<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
    <meta name="keywords" content="Hexo Theme Keep">
    <meta name="description" content="Hexo Theme Keep">
    <meta name="author" content="Blank">
    
    <title>
        
            RocketMQ-01 |
        
        Blankの博客
    </title>
    
<link rel="stylesheet" href="/css/style.css">

    <link rel="shortcut icon" href="/images/logo.jpg">
    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/font/css/fontawesome.min.css">
    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/font/css/regular.min.css">
    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/font/css/solid.min.css">
    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/font/css/brands.min.css">
    <script id="hexo-configurations">
    let KEEP = window.KEEP || {}
    KEEP.hexo_config = {"hostname":"example.com","root":"/","language":"zh-CN","path":"search.json"}
    KEEP.theme_config = {"toc":{"enable":true,"number":true,"expand_all":true,"init_open":true},"style":{"primary_color":"#0066cc","logo":"/images/logo.jpg","favicon":"/images/logo.jpg","avatar":"/images/logo.jpg","font_size":"18px","font_family":"STKaiti","hover":{"shadow":true,"scale":true},"first_screen":{"enable":true,"header_transparent":true,"background_img":"/images/bg.svg","description":"Keep writing and Keep loving.","font_color":null,"hitokoto":true},"scroll":{"progress_bar":true,"percent":true}},"local_search":{"enable":true,"preload":true},"code_copy":{},"code_block":{"tools":{"enable":true,"style":"mac"},"highlight_theme":"default"},"side_tools":{},"pjax":{"enable":true},"lazyload":{"enable":true},"comment":{"enable":false,"use":"gitalk","valine":{"appid":null,"appkey":null,"server_urls":null,"placeholder":null},"gitalk":{"github_id":"5ober","github_admins":"5ober","repository":"hexo-blog-comments","client_id":"40d85a7b36388c0b5094","client_secret":"6c7eab92d24b6f1bdfbcdf077c73e86841b35d5e","proxy":null},"twikoo":{"env_id":null,"region":null,"version":"1.6.8"},"waline":{"server_url":null,"reaction":false,"version":2}},"post":{"author_label":{"enable":true,"auto":false,"custom_label_list":["Trainee","Engineer","Java工程师"]},"word_count":{"enable":true,"wordcount":true,"min2read":true},"img_align":"left","copyright_info":true},"version":"3.6.1"}
    KEEP.language_ago = {"second":"%s 秒前","minute":"%s 分钟前","hour":"%s 小时前","day":"%s 天前","week":"%s 周前","month":"%s 个月前","year":"%s 年前"}
    KEEP.language_code_block = {"copy":"复制代码","copied":"已复制","fold":"折叠代码块","folded":"已折叠"}
    KEEP.language_copy_copyright = {"copy":"复制版权信息","copied":"已复制","title":"原文标题","author":"原文作者","link":"原文链接"}
  </script>
<meta name="generator" content="Hexo 6.3.0"><link rel="alternate" href="/atom.xml" title="Blankの博客" type="application/atom+xml">
</head>


<body>
<div class="progress-bar-container">
    
        <span class="scroll-progress-bar"></span>
    

    
        <span class="pjax-progress-bar"></span>
        <i class="pjax-progress-icon fas fa-circle-notch fa-spin"></i>
    
</div>


<main class="page-container">

    

    <div class="page-main-content">

        <div class="page-main-content-top">
            
<header class="header-wrapper">

    <div class="header-content">
        <div class="left">
            
                <a class="logo-image" href="/">
                    <img src="/images/logo.jpg">
                </a>
            
            <a class="logo-title" href="/">
               Blankの博客
            </a>
        </div>

        <div class="right">
            <div class="pc">
                <ul class="menu-list">
                    
                        <li class="menu-item">
                            <a class=""
                               href="/"
                            >
                                首页
                            </a>
                        </li>
                    
                        <li class="menu-item">
                            <a class=""
                               href="/archives"
                            >
                                归档
                            </a>
                        </li>
                    
                        <li class="menu-item">
                            <a class=""
                               href="/tags"
                            >
                                标签
                            </a>
                        </li>
                    
                        <li class="menu-item">
                            <a class=""
                               href="/categories"
                            >
                                分类
                            </a>
                        </li>
                    
                        <li class="menu-item">
                            <a class=""
                               href="/links"
                            >
                                友链
                            </a>
                        </li>
                    
                        <li class="menu-item">
                            <a class=""
                               href="/about"
                            >
                                关于
                            </a>
                        </li>
                    
                    
                        <li class="menu-item search search-popup-trigger">
                            <i class="fas fa-search"></i>
                        </li>
                    
                </ul>
            </div>
            <div class="mobile">
                
                    <div class="icon-item search search-popup-trigger"><i class="fas fa-search"></i></div>
                
                <div class="icon-item menu-bar">
                    <div class="menu-bar-middle"></div>
                </div>
            </div>
        </div>
    </div>

    <div class="header-drawer">
        <ul class="drawer-menu-list">
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/">首页</a>
                </li>
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/archives">归档</a>
                </li>
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/tags">标签</a>
                </li>
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/categories">分类</a>
                </li>
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/links">友链</a>
                </li>
            
                <li class="drawer-menu-item flex-center">
                    <a class=""
                       href="/about">关于</a>
                </li>
            
        </ul>
    </div>

    <div class="window-mask"></div>

</header>


        </div>

        <div class="page-main-content-middle">

            <div class="main-content">

                
                    <div class="fade-in-down-animation">
    <div class="post-page-container">
        <div class="article-content-container">

            <div class="article-title">
                <span class="title-hover-animation">RocketMQ-01</span>
            </div>

            
                <div class="article-header">
                    <div class="avatar">
                        <img src="/images/logo.jpg">
                    </div>
                    <div class="info">
                        <div class="author">
                            <span class="name">Blank</span>
                            
                                <span class="author-label">Java工程师</span>
                            
                        </div>
                        <div class="meta-info">
                            
<div class="article-meta-info">
    <span class="article-date article-meta-item">
        
            <i class="fa-regular fa-calendar-plus"></i>&nbsp;
        
        <span class="pc">2023-02-19 00:00:00</span>
        <span class="mobile">2023-02-19 00:00</span>
    </span>
    
        <span class="article-update-date article-meta-item">
        <i class="fas fa-file-pen"></i>&nbsp;
        <span class="pc">2023-03-22 10:17:19</span>
    </span>
    
    
        <span class="article-categories article-meta-item">
            <i class="fas fa-folder"></i>&nbsp;
            <ul>
                
                    <li>
                        <a href="/categories/RocketMQ/">RocketMQ</a>&nbsp;
                    </li>
                
            </ul>
        </span>
    
    
        <span class="article-tags article-meta-item">
            <i class="fas fa-tags"></i>&nbsp;
            <ul>
                
                    <li>
                        <a href="/tags/%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6/">消息中间件</a>&nbsp;
                    </li>
                
            </ul>
        </span>
    

    
    
        <span class="article-wordcount article-meta-item">
            <i class="fas fa-file-word"></i>&nbsp;<span>12.7k 字</span>
        </span>
    
    
        <span class="article-min2read article-meta-item">
            <i class="fas fa-clock"></i>&nbsp;<span>54 分钟</span>
        </span>
    
    
        <span class="article-pv article-meta-item">
            <i class="fas fa-eye"></i>&nbsp;<span id="busuanzi_value_page_pv"></span>
        </span>
    
</div>

                        </div>
                    </div>
                </div>
            

            <div class="article-content keep-markdown-body">
                

                <h1 id="RocketMQ-01"><a href="#RocketMQ-01" class="headerlink" title="RocketMQ-01"></a>RocketMQ-01</h1><ul>
<li>1.2 MQ的优点和缺点</li>
<li>1.3 各种MQ产品的比较</li>
<li>2.1 准备工作<ul>
<li>2.1.1 下载RocketMQ</li>
<li>2.2.2 环境要求</li>
</ul>
</li>
<li>2.2 安装RocketMQ<ul>
<li>2.2.1 安装步骤</li>
<li>2.2.2 目录介绍</li>
</ul>
</li>
<li>2.3 启动RocketMQ</li>
<li>2.4 测试RocketMQ<ul>
<li>2.4.1 发送消息</li>
<li>2.4.2 接收消息</li>
</ul>
</li>
<li>2.5 关闭RocketMQ</li>
<li>3.1 各角色介绍</li>
<li>3.2 集群搭建方式<ul>
<li>3.2.1 集群特点</li>
<li>3.2.3 集群模式</li>
</ul>
</li>
<li>3.3 双主双从集群搭建<ul>
<li>3.3.1 总体架构</li>
<li>3.3.2 集群工作流程</li>
<li>3.3.3 服务器环境</li>
<li>3.3.4 Host添加信息</li>
<li>3.3.5 防火墙配置</li>
<li>3.3.6 环境变量配置</li>
<li>3.3.7 创建消息存储路径</li>
<li>3.3.8 broker配置文件</li>
<li>3.3.9 修改启动脚本文件</li>
<li>3.3.10 服务启动</li>
<li>3.3.11 查看进程状态</li>
<li>3.3.12 查看日志</li>
</ul>
</li>
<li>3.4 mqadmin管理工具<ul>
<li>3.4.1 使用方式</li>
<li>3.4.3 注意事项</li>
</ul>
</li>
<li>3.5 集群监控平台搭建<ul>
<li>3.5.1 概述</li>
<li>3.5.2 下载并编译打包</li>
</ul>
</li>
<li>4.1 基本样例<ul>
<li>4.1.1 消息发送</li>
<li>4.1.2 消费消息</li>
</ul>
</li>
<li>4.2 顺序消息<ul>
<li>4.2.1 顺序消息生产</li>
<li>4.2.2 顺序消费消息</li>
</ul>
</li>
<li>4.3 延时消息<ul>
<li>4.3.1 启动消息消费者</li>
<li>4.3.2 发送延时消息</li>
<li>4.3.4 使用限制</li>
</ul>
</li>
<li>4.4 批量消息<ul>
<li>4.4.1 发送批量消息</li>
</ul>
</li>
<li>4.5 过滤消息<ul>
<li>4.5.1 SQL基本语法</li>
<li>4.5.2 消息生产者</li>
<li>4.5.3 消息消费者</li>
</ul>
</li>
<li>4.6 事务消息<ul>
<li>4.6.2 使用限制</li>
</ul>
</li>
</ul>
<h1 id="1-MQ介绍"><a href="#1-MQ介绍" class="headerlink" title="1. MQ介绍"></a>1. MQ介绍</h1><p>##1.1 为什么要用MQ</p>
<p>消息队列是一种“先进先出”的数据结构</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/queue1.png"
                      alt="img"
                ></p>
<p>其应用场景主要包含以下3个方面</p>
<ul>
<li>应用解耦</li>
</ul>
<p>系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/%E8%A7%A3%E8%80%A61.png"
                      alt="img"
                ></p>
<p>使用消息队列解耦合，系统的耦合性就会提高了。比如物流系统发生故障，需要几分钟才能来修复，在这段时间内，物流系统要处理的数据被缓存到消息队列中，用户的下单操作正常完成。当物流系统回复后，补充处理存在消息队列中的订单消息即可，终端系统感知不到物流系统发生过几分钟故障。</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/%E8%A7%A3%E8%80%A62.png"
                      alt="img"
                ></p>
<ul>
<li>流量削峰</li>
</ul>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/mq-5.png"
                      alt="img"
                ></p>
<p>应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/mq-6.png"
                      alt="img"
                ></p>
<p>一般情况，为了保证系统的稳定性，如果系统负载超过阈值，就会阻止用户请求，这会影响用户体验，而如果使用消息队列将请求缓存起来，等待系统处理完毕后通知用户下单完毕，这样总不能下单体验要好。</p>
<p>处于经济考量目的：</p>
<p>业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显然不划算，这时可以使用消息队列对峰值流量削峰</p>
<ul>
<li>数据分发</li>
</ul>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/mq-1.png"
                      alt="img"
                ></p>
<p>通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/mq-2.png"
                      alt="img"
                ></p>
<h2 id="1-2-MQ的优点和缺点"><a href="#1-2-MQ的优点和缺点" class="headerlink" title="1.2 MQ的优点和缺点"></a>1.2 MQ的优点和缺点</h2><p>优点：解耦、削峰、数据分发</p>
<p>缺点包含以下几点：</p>
<ul>
<li><p>系统可用性降低</p>
<p>系统引入的外部依赖越多，系统稳定性越差。一旦MQ宕机，就会对业务造成影响。</p>
<p>如何保证MQ的高可用？</p>
</li>
<li><p>系统复杂度提高</p>
<p>MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。</p>
<p>如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？</p>
</li>
<li><p>一致性问题</p>
<p>A系统处理完业务，通过MQ给B、C、D三个系统发消息数据，如果B系统、C系统处理成功，D系统处理失败。</p>
<p>如何保证消息数据处理的一致性？</p>
</li>
</ul>
<h2 id="1-3-各种MQ产品的比较"><a href="#1-3-各种MQ产品的比较" class="headerlink" title="1.3 各种MQ产品的比较"></a>1.3 各种MQ产品的比较</h2><p>常见的MQ产品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ。</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/MQ%E6%AF%94%E8%BE%83.png"
                      alt="img"
                ></p>
<h1 id="2-RocketMQ快速入门"><a href="#2-RocketMQ快速入门" class="headerlink" title="2. RocketMQ快速入门"></a>2. RocketMQ快速入门</h1><p>RocketMQ是阿里巴巴2016年MQ中间件，使用Java语言开发，在阿里内部，RocketMQ承接了例如“双11”等高并发场景的消息流转，能够处理万亿级别的消息。</p>
<h2 id="2-1-准备工作"><a href="#2-1-准备工作" class="headerlink" title="2.1 准备工作"></a>2.1 准备工作</h2><h3 id="2-1-1-下载RocketMQ"><a href="#2-1-1-下载RocketMQ" class="headerlink" title="2.1.1 下载RocketMQ"></a>2.1.1 下载RocketMQ</h3><p>RocketMQ最新版本：4.5.1</p>
<p>下载地址</p>
<h3 id="2-2-2-环境要求"><a href="#2-2-2-环境要求" class="headerlink" title="2.2.2 环境要求"></a>2.2.2 环境要求</h3><ul>
<li>Linux64位系统</li>
<li>JDK1.8(64位)</li>
<li>源码安装需要安装Maven 3.2.x</li>
</ul>
<h2 id="2-2-安装RocketMQ"><a href="#2-2-安装RocketMQ" class="headerlink" title="2.2 安装RocketMQ"></a>2.2 安装RocketMQ</h2><h3 id="2-2-1-安装步骤"><a href="#2-2-1-安装步骤" class="headerlink" title="2.2.1 安装步骤"></a>2.2.1 安装步骤</h3><p>本教程以二进制包方式安装</p>
<ol>
<li>解压安装包</li>
<li>进入安装目录</li>
</ol>
<h3 id="2-2-2-目录介绍"><a href="#2-2-2-目录介绍" class="headerlink" title="2.2.2 目录介绍"></a>2.2.2 目录介绍</h3><ul>
<li>bin：启动脚本，包括shell脚本和CMD脚本</li>
<li>conf：实例配置文件 ，包括broker配置文件、logback配置文件等</li>
<li>lib：依赖jar包，包括Netty、commons-lang、FastJSON等</li>
</ul>
<h2 id="2-3-启动RocketMQ"><a href="#2-3-启动RocketMQ" class="headerlink" title="2.3 启动RocketMQ"></a>2.3 启动RocketMQ</h2><ol>
<li>启动NameServer</li>
</ol>
<figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta prompt_"># </span><span class="language-bash">1.启动NameServer</span></span><br><span class="line">nohup sh bin/mqnamesrv &amp;</span><br><span class="line"><span class="meta prompt_"># </span><span class="language-bash">2.查看启动日志</span></span><br><span class="line">tail -f ~/logs/rocketmqlogs/namesrv.log</span><br></pre></td></tr></table></figure>



<ol>
<li>启动Broker</li>
</ol>
<figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta prompt_"># </span><span class="language-bash">1.启动Broker</span></span><br><span class="line">nohup sh bin/mqbroker -n localhost:9876 &amp;</span><br><span class="line"><span class="meta prompt_"># </span><span class="language-bash">2.查看启动日志</span></span><br><span class="line">tail -f ~/logs/rocketmqlogs/broker.log </span><br></pre></td></tr></table></figure>



<ul>
<li><p>问题描述：</p>
<p>RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小</p>
</li>
</ul>
<figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta prompt_"># </span><span class="language-bash">编辑runbroker.sh和runserver.sh修改默认JVM大小</span></span><br><span class="line">vi runbroker.sh</span><br><span class="line">vi runserver.sh</span><br></pre></td></tr></table></figure>



<ul>
<li>参考设置：</li>
</ul>
<figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">JAVA_OPT=&quot;$&#123;JAVA_OPT&#125; -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m&quot;</span><br></pre></td></tr></table></figure>

<h2 id="2-4-测试RocketMQ"><a href="#2-4-测试RocketMQ" class="headerlink" title="2.4 测试RocketMQ"></a>2.4 测试RocketMQ</h2><h3 id="2-4-1-发送消息"><a href="#2-4-1-发送消息" class="headerlink" title="2.4.1 发送消息"></a>2.4.1 发送消息</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># 1.设置环境变量</span></span><br><span class="line"><span class="built_in">export</span> NAMESRV_ADDR=localhost:9876</span><br><span class="line"><span class="comment"># 2.使用安装包的Demo发送消息</span></span><br><span class="line">sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer</span><br></pre></td></tr></table></figure>



<h3 id="2-4-2-接收消息"><a href="#2-4-2-接收消息" class="headerlink" title="2.4.2 接收消息"></a>2.4.2 接收消息</h3><figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta prompt_"># </span><span class="language-bash">1.设置环境变量</span></span><br><span class="line">export NAMESRV_ADDR=localhost:9876</span><br><span class="line"><span class="meta prompt_"># </span><span class="language-bash">2.接收消息</span></span><br><span class="line">sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer</span><br></pre></td></tr></table></figure>



<h2 id="2-5-关闭RocketMQ"><a href="#2-5-关闭RocketMQ" class="headerlink" title="2.5 关闭RocketMQ"></a>2.5 关闭RocketMQ</h2><figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta prompt_"># </span><span class="language-bash">1.关闭NameServer</span></span><br><span class="line">sh bin/mqshutdown namesrv</span><br><span class="line"><span class="meta prompt_"># </span><span class="language-bash">2.关闭Broker</span></span><br><span class="line">sh bin/mqshutdown broker</span><br></pre></td></tr></table></figure>



<h1 id="3-RocketMQ集群搭建"><a href="#3-RocketMQ集群搭建" class="headerlink" title="3. RocketMQ集群搭建"></a>3. RocketMQ集群搭建</h1><h2 id="3-1-各角色介绍"><a href="#3-1-各角色介绍" class="headerlink" title="3.1 各角色介绍"></a>3.1 各角色介绍</h2><ul>
<li>Producer：消息的发送者；举例：发信者</li>
<li>Consumer：消息接收者；举例：收信者</li>
<li>Broker：暂存和传输消息；举例：邮局</li>
<li>NameServer：管理Broker；举例：各个邮局的管理机构</li>
<li>Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息</li>
<li>Message Queue：相当于是Topic的分区；用于并行发送和接收消息</li>
</ul>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/RocketMQ%E8%A7%92%E8%89%B2.jpg"
                      alt="img"
                ></p>
<h2 id="3-2-集群搭建方式"><a href="#3-2-集群搭建方式" class="headerlink" title="3.2 集群搭建方式"></a>3.2 集群搭建方式</h2><h3 id="3-2-1-集群特点"><a href="#3-2-1-集群特点" class="headerlink" title="3.2.1 集群特点"></a>3.2.1 集群特点</h3><ul>
<li>NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。</li>
<li>Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。</li>
<li>Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。</li>
<li>Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。</li>
</ul>
<h3 id="3-2-3-集群模式"><a href="#3-2-3-集群模式" class="headerlink" title="3.2.3 集群模式"></a>3.2.3 集群模式</h3><h4 id="1）单Master模式"><a href="#1）单Master模式" class="headerlink" title="1）单Master模式"></a>1）单Master模式</h4><p>这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。</p>
<h4 id="2）多Master模式"><a href="#2）多Master模式" class="headerlink" title="2）多Master模式"></a>2）多Master模式</h4><p>一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：</p>
<ul>
<li>优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；</li>
<li>缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。</li>
</ul>
<h4 id="3）多Master多Slave模式（异步）"><a href="#3）多Master多Slave模式（异步）" class="headerlink" title="3）多Master多Slave模式（异步）"></a>3）多Master多Slave模式（异步）</h4><p>每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：</p>
<ul>
<li>优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；</li>
<li>缺点：Master宕机，磁盘损坏情况下会丢失少量消息。</li>
</ul>
<h4 id="4）多Master多Slave模式（同步）"><a href="#4）多Master多Slave模式（同步）" class="headerlink" title="4）多Master多Slave模式（同步）"></a>4）多Master多Slave模式（同步）</h4><p>每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：</p>
<ul>
<li>优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；</li>
<li>缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。</li>
</ul>
<h2 id="3-3-双主双从集群搭建"><a href="#3-3-双主双从集群搭建" class="headerlink" title="3.3 双主双从集群搭建"></a>3.3 双主双从集群搭建</h2><h3 id="3-3-1-总体架构"><a href="#3-3-1-总体架构" class="headerlink" title="3.3.1 总体架构"></a>3.3.1 总体架构</h3><p>消息高可用采用2m-2s（同步双写）方式</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/RocketMQ%E9%9B%86%E7%BE%A4.png"
                      alt="img"
                ></p>
<h3 id="3-3-2-集群工作流程"><a href="#3-3-2-集群工作流程" class="headerlink" title="3.3.2 集群工作流程"></a>3.3.2 集群工作流程</h3><ol>
<li>启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。</li>
<li>Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。</li>
<li>收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。</li>
<li>Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。</li>
<li>Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。</li>
</ol>
<h3 id="3-3-3-服务器环境"><a href="#3-3-3-服务器环境" class="headerlink" title="3.3.3 服务器环境"></a>3.3.3 服务器环境</h3><table>
<thead>
<tr>
<th><strong>序号</strong></th>
<th><strong>IP</strong></th>
<th><strong>角色</strong></th>
<th><strong>架构模式</strong></th>
</tr>
</thead>
<tbody><tr>
<td>1</td>
<td>192.168.25.135</td>
<td>nameserver、brokerserver</td>
<td>Master1、Slave2</td>
</tr>
<tr>
<td>2</td>
<td>192.168.25.138</td>
<td>nameserver、brokerserver</td>
<td>Master2、Slave1</td>
</tr>
</tbody></table>
<h3 id="3-3-4-Host添加信息"><a href="#3-3-4-Host添加信息" class="headerlink" title="3.3.4 Host添加信息"></a>3.3.4 Host添加信息</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim /etc/hosts</span><br></pre></td></tr></table></figure>



<p>配置如下:</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># nameserver</span></span><br><span class="line">192.168.25.135 rocketmq-nameserver1</span><br><span class="line">192.168.25.138 rocketmq-nameserver2</span><br><span class="line"><span class="comment"># broker</span></span><br><span class="line">192.168.25.135 rocketmq-master1</span><br><span class="line">192.168.25.135 rocketmq-slave2</span><br><span class="line">192.168.25.138 rocketmq-master2</span><br><span class="line">192.168.25.138 rocketmq-slave1</span><br></pre></td></tr></table></figure>



<p>配置完成后, 重启网卡</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">systemctl restart network</span><br></pre></td></tr></table></figure>



<h3 id="3-3-5-防火墙配置"><a href="#3-3-5-防火墙配置" class="headerlink" title="3.3.5 防火墙配置"></a>3.3.5 防火墙配置</h3><p>宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># 关闭防火墙</span></span><br><span class="line">systemctl stop firewalld.service </span><br><span class="line"><span class="comment"># 查看防火墙的状态</span></span><br><span class="line">firewall-cmd --state </span><br><span class="line"><span class="comment"># 禁止firewall开机启动</span></span><br><span class="line">systemctl <span class="built_in">disable</span> firewalld.service</span><br></pre></td></tr></table></figure>



<p>或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：</p>
<ul>
<li><code>nameserver</code> 默认使用 9876 端口</li>
<li><code>master</code> 默认使用 10911 端口</li>
<li><code>slave</code> 默认使用11011 端口</li>
</ul>
<p>执行以下命令：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># 开放name server默认端口</span></span><br><span class="line">firewall-cmd --remove-port=9876/tcp --permanent</span><br><span class="line"><span class="comment"># 开放master默认端口</span></span><br><span class="line">firewall-cmd --remove-port=10911/tcp --permanent</span><br><span class="line"><span class="comment"># 开放slave默认端口 (当前集群模式可不开启)</span></span><br><span class="line">firewall-cmd --remove-port=11011/tcp --permanent </span><br><span class="line"><span class="comment"># 重启防火墙</span></span><br><span class="line">firewall-cmd --reload</span><br></pre></td></tr></table></figure>



<h3 id="3-3-6-环境变量配置"><a href="#3-3-6-环境变量配置" class="headerlink" title="3.3.6 环境变量配置"></a>3.3.6 环境变量配置</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim /etc/profile</span><br></pre></td></tr></table></figure>



<p>在profile文件的末尾加入如下命令</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#set rocketmq</span></span><br><span class="line">ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-release</span><br><span class="line">PATH=<span class="variable">$PATH</span>:<span class="variable">$ROCKETMQ_HOME</span>/bin</span><br><span class="line"><span class="built_in">export</span> ROCKETMQ_HOME PATH</span><br></pre></td></tr></table></figure>



<p>输入:wq! 保存并退出， 并使得配置立刻生效：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">source</span> /etc/profile</span><br></pre></td></tr></table></figure>



<h3 id="3-3-7-创建消息存储路径"><a href="#3-3-7-创建消息存储路径" class="headerlink" title="3.3.7 创建消息存储路径"></a>3.3.7 创建消息存储路径</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">mkdir</span> /usr/local/rocketmq/store</span><br><span class="line"><span class="built_in">mkdir</span> /usr/local/rocketmq/store/commitlog</span><br><span class="line"><span class="built_in">mkdir</span> /usr/local/rocketmq/store/consumequeue</span><br><span class="line"><span class="built_in">mkdir</span> /usr/local/rocketmq/store/index</span><br></pre></td></tr></table></figure>



<h3 id="3-3-8-broker配置文件"><a href="#3-3-8-broker配置文件" class="headerlink" title="3.3.8 broker配置文件"></a>3.3.8 broker配置文件</h3><h4 id="1）master1"><a href="#1）master1" class="headerlink" title="1）master1"></a>1）master1</h4><p>服务器：192.168.25.135</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vi /usr/soft/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-a.properties</span><br></pre></td></tr></table></figure>



<p>修改配置如下：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#所属集群名字</span></span><br><span class="line">brokerClusterName=rocketmq-cluster</span><br><span class="line"><span class="comment">#broker名字，注意此处不同的配置文件填写的不一样</span></span><br><span class="line">brokerName=broker-a</span><br><span class="line"><span class="comment">#0 表示 Master，&gt;0 表示 Slave</span></span><br><span class="line">brokerId=0</span><br><span class="line"><span class="comment">#nameServer地址，分号分割</span></span><br><span class="line">namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876</span><br><span class="line"><span class="comment">#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数</span></span><br><span class="line">defaultTopicQueueNums=4</span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateTopicEnable=<span class="literal">true</span></span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateSubscriptionGroup=<span class="literal">true</span></span><br><span class="line"><span class="comment">#Broker 对外服务的监听端口</span></span><br><span class="line">listenPort=10911</span><br><span class="line"><span class="comment">#删除文件时间点，默认凌晨 4点</span></span><br><span class="line">deleteWhen=04</span><br><span class="line"><span class="comment">#文件保留时间，默认 48 小时</span></span><br><span class="line">fileReservedTime=120</span><br><span class="line"><span class="comment">#commitLog每个文件的大小默认1G</span></span><br><span class="line">mapedFileSizeCommitLog=1073741824</span><br><span class="line"><span class="comment">#ConsumeQueue每个文件默认存30W条，根据业务情况调整</span></span><br><span class="line">mapedFileSizeConsumeQueue=300000</span><br><span class="line"><span class="comment">#destroyMapedFileIntervalForcibly=120000</span></span><br><span class="line"><span class="comment">#redeleteHangedFileInterval=120000</span></span><br><span class="line"><span class="comment">#检测物理文件磁盘空间</span></span><br><span class="line">diskMaxUsedSpaceRatio=88</span><br><span class="line"><span class="comment">#存储路径</span></span><br><span class="line">storePathRootDir=/usr/local/rocketmq/store</span><br><span class="line"><span class="comment">#commitLog 存储路径</span></span><br><span class="line">storePathCommitLog=/usr/local/rocketmq/store/commitlog</span><br><span class="line"><span class="comment">#消费队列存储路径存储路径</span></span><br><span class="line">storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue</span><br><span class="line"><span class="comment">#消息索引存储路径</span></span><br><span class="line">storePathIndex=/usr/local/rocketmq/store/index</span><br><span class="line"><span class="comment">#checkpoint 文件存储路径</span></span><br><span class="line">storeCheckpoint=/usr/local/rocketmq/store/checkpoint</span><br><span class="line"><span class="comment">#abort 文件存储路径</span></span><br><span class="line">abortFile=/usr/local/rocketmq/store/abort</span><br><span class="line"><span class="comment">#限制的消息大小</span></span><br><span class="line">maxMessageSize=65536</span><br><span class="line"><span class="comment">#flushCommitLogLeastPages=4</span></span><br><span class="line"><span class="comment">#flushConsumeQueueLeastPages=2</span></span><br><span class="line"><span class="comment">#flushCommitLogThoroughInterval=10000</span></span><br><span class="line"><span class="comment">#flushConsumeQueueThoroughInterval=60000</span></span><br><span class="line"><span class="comment">#Broker 的角色</span></span><br><span class="line"><span class="comment">#- ASYNC_MASTER 异步复制Master</span></span><br><span class="line"><span class="comment">#- SYNC_MASTER 同步双写Master</span></span><br><span class="line"><span class="comment">#- SLAVE</span></span><br><span class="line">brokerRole=SYNC_MASTER</span><br><span class="line"><span class="comment">#刷盘方式</span></span><br><span class="line"><span class="comment">#- ASYNC_FLUSH 异步刷盘</span></span><br><span class="line"><span class="comment">#- SYNC_FLUSH 同步刷盘</span></span><br><span class="line">flushDiskType=SYNC_FLUSH</span><br><span class="line"><span class="comment">#checkTransactionMessageEnable=false</span></span><br><span class="line"><span class="comment">#发消息线程池数量</span></span><br><span class="line"><span class="comment">#sendMessageThreadPoolNums=128</span></span><br><span class="line"><span class="comment">#拉消息线程池数量</span></span><br><span class="line"><span class="comment">#pullMessageThreadPoolNums=128</span></span><br></pre></td></tr></table></figure>



<h4 id="2）slave2"><a href="#2）slave2" class="headerlink" title="2）slave2"></a>2）slave2</h4><p>服务器：192.168.25.135</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vi /usr/soft/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-b-s.properties</span><br></pre></td></tr></table></figure>



<p>修改配置如下：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#所属集群名字</span></span><br><span class="line">brokerClusterName=rocketmq-cluster</span><br><span class="line"><span class="comment">#broker名字，注意此处不同的配置文件填写的不一样</span></span><br><span class="line">brokerName=broker-b</span><br><span class="line"><span class="comment">#0 表示 Master，&gt;0 表示 Slave</span></span><br><span class="line">brokerId=1</span><br><span class="line"><span class="comment">#nameServer地址，分号分割</span></span><br><span class="line">namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876</span><br><span class="line"><span class="comment">#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数</span></span><br><span class="line">defaultTopicQueueNums=4</span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateTopicEnable=<span class="literal">true</span></span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateSubscriptionGroup=<span class="literal">true</span></span><br><span class="line"><span class="comment">#Broker 对外服务的监听端口</span></span><br><span class="line">listenPort=11011</span><br><span class="line"><span class="comment">#删除文件时间点，默认凌晨 4点</span></span><br><span class="line">deleteWhen=04</span><br><span class="line"><span class="comment">#文件保留时间，默认 48 小时</span></span><br><span class="line">fileReservedTime=120</span><br><span class="line"><span class="comment">#commitLog每个文件的大小默认1G</span></span><br><span class="line">mapedFileSizeCommitLog=1073741824</span><br><span class="line"><span class="comment">#ConsumeQueue每个文件默认存30W条，根据业务情况调整</span></span><br><span class="line">mapedFileSizeConsumeQueue=300000</span><br><span class="line"><span class="comment">#destroyMapedFileIntervalForcibly=120000</span></span><br><span class="line"><span class="comment">#redeleteHangedFileInterval=120000</span></span><br><span class="line"><span class="comment">#检测物理文件磁盘空间</span></span><br><span class="line">diskMaxUsedSpaceRatio=88</span><br><span class="line"><span class="comment">#存储路径</span></span><br><span class="line">storePathRootDir=/usr/local/rocketmq/store</span><br><span class="line"><span class="comment">#commitLog 存储路径</span></span><br><span class="line">storePathCommitLog=/usr/local/rocketmq/store/commitlog</span><br><span class="line"><span class="comment">#消费队列存储路径存储路径</span></span><br><span class="line">storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue</span><br><span class="line"><span class="comment">#消息索引存储路径</span></span><br><span class="line">storePathIndex=/usr/local/rocketmq/store/index</span><br><span class="line"><span class="comment">#checkpoint 文件存储路径</span></span><br><span class="line">storeCheckpoint=/usr/local/rocketmq/store/checkpoint</span><br><span class="line"><span class="comment">#abort 文件存储路径</span></span><br><span class="line">abortFile=/usr/local/rocketmq/store/abort</span><br><span class="line"><span class="comment">#限制的消息大小</span></span><br><span class="line">maxMessageSize=65536</span><br><span class="line"><span class="comment">#flushCommitLogLeastPages=4</span></span><br><span class="line"><span class="comment">#flushConsumeQueueLeastPages=2</span></span><br><span class="line"><span class="comment">#flushCommitLogThoroughInterval=10000</span></span><br><span class="line"><span class="comment">#flushConsumeQueueThoroughInterval=60000</span></span><br><span class="line"><span class="comment">#Broker 的角色</span></span><br><span class="line"><span class="comment">#- ASYNC_MASTER 异步复制Master</span></span><br><span class="line"><span class="comment">#- SYNC_MASTER 同步双写Master</span></span><br><span class="line"><span class="comment">#- SLAVE</span></span><br><span class="line">brokerRole=SLAVE</span><br><span class="line"><span class="comment">#刷盘方式</span></span><br><span class="line"><span class="comment">#- ASYNC_FLUSH 异步刷盘</span></span><br><span class="line"><span class="comment">#- SYNC_FLUSH 同步刷盘</span></span><br><span class="line">flushDiskType=ASYNC_FLUSH</span><br><span class="line"><span class="comment">#checkTransactionMessageEnable=false</span></span><br><span class="line"><span class="comment">#发消息线程池数量</span></span><br><span class="line"><span class="comment">#sendMessageThreadPoolNums=128</span></span><br><span class="line"><span class="comment">#拉消息线程池数量</span></span><br><span class="line"><span class="comment">#pullMessageThreadPoolNums=128</span></span><br></pre></td></tr></table></figure>




<h4 id="3）master2"><a href="#3）master2" class="headerlink" title="3）master2"></a>3）master2</h4><p>服务器：192.168.25.138</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vi /usr/soft/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-b.properties</span><br></pre></td></tr></table></figure>



<p>修改配置如下：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#所属集群名字</span></span><br><span class="line">brokerClusterName=rocketmq-cluster</span><br><span class="line"><span class="comment">#broker名字，注意此处不同的配置文件填写的不一样</span></span><br><span class="line">brokerName=broker-b</span><br><span class="line"><span class="comment">#0 表示 Master，&gt;0 表示 Slave</span></span><br><span class="line">brokerId=0</span><br><span class="line"><span class="comment">#nameServer地址，分号分割</span></span><br><span class="line">namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876</span><br><span class="line"><span class="comment">#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数</span></span><br><span class="line">defaultTopicQueueNums=4</span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateTopicEnable=<span class="literal">true</span></span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateSubscriptionGroup=<span class="literal">true</span></span><br><span class="line"><span class="comment">#Broker 对外服务的监听端口</span></span><br><span class="line">listenPort=10911</span><br><span class="line"><span class="comment">#删除文件时间点，默认凌晨 4点</span></span><br><span class="line">deleteWhen=04</span><br><span class="line"><span class="comment">#文件保留时间，默认 48 小时</span></span><br><span class="line">fileReservedTime=120</span><br><span class="line"><span class="comment">#commitLog每个文件的大小默认1G</span></span><br><span class="line">mapedFileSizeCommitLog=1073741824</span><br><span class="line"><span class="comment">#ConsumeQueue每个文件默认存30W条，根据业务情况调整</span></span><br><span class="line">mapedFileSizeConsumeQueue=300000</span><br><span class="line"><span class="comment">#destroyMapedFileIntervalForcibly=120000</span></span><br><span class="line"><span class="comment">#redeleteHangedFileInterval=120000</span></span><br><span class="line"><span class="comment">#检测物理文件磁盘空间</span></span><br><span class="line">diskMaxUsedSpaceRatio=88</span><br><span class="line"><span class="comment">#存储路径</span></span><br><span class="line">storePathRootDir=/usr/local/rocketmq/store</span><br><span class="line"><span class="comment">#commitLog 存储路径</span></span><br><span class="line">storePathCommitLog=/usr/local/rocketmq/store/commitlog</span><br><span class="line"><span class="comment">#消费队列存储路径存储路径</span></span><br><span class="line">storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue</span><br><span class="line"><span class="comment">#消息索引存储路径</span></span><br><span class="line">storePathIndex=/usr/local/rocketmq/store/index</span><br><span class="line"><span class="comment">#checkpoint 文件存储路径</span></span><br><span class="line">storeCheckpoint=/usr/local/rocketmq/store/checkpoint</span><br><span class="line"><span class="comment">#abort 文件存储路径</span></span><br><span class="line">abortFile=/usr/local/rocketmq/store/abort</span><br><span class="line"><span class="comment">#限制的消息大小</span></span><br><span class="line">maxMessageSize=65536</span><br><span class="line"><span class="comment">#flushCommitLogLeastPages=4</span></span><br><span class="line"><span class="comment">#flushConsumeQueueLeastPages=2</span></span><br><span class="line"><span class="comment">#flushCommitLogThoroughInterval=10000</span></span><br><span class="line"><span class="comment">#flushConsumeQueueThoroughInterval=60000</span></span><br><span class="line"><span class="comment">#Broker 的角色</span></span><br><span class="line"><span class="comment">#- ASYNC_MASTER 异步复制Master</span></span><br><span class="line"><span class="comment">#- SYNC_MASTER 同步双写Master</span></span><br><span class="line"><span class="comment">#- SLAVE</span></span><br><span class="line">brokerRole=SYNC_MASTER</span><br><span class="line"><span class="comment">#刷盘方式</span></span><br><span class="line"><span class="comment">#- ASYNC_FLUSH 异步刷盘</span></span><br><span class="line"><span class="comment">#- SYNC_FLUSH 同步刷盘</span></span><br><span class="line">flushDiskType=SYNC_FLUSH</span><br><span class="line"><span class="comment">#checkTransactionMessageEnable=false</span></span><br><span class="line"><span class="comment">#发消息线程池数量</span></span><br><span class="line"><span class="comment">#sendMessageThreadPoolNums=128</span></span><br><span class="line"><span class="comment">#拉消息线程池数量</span></span><br><span class="line"><span class="comment">#pullMessageThreadPoolNums=128</span></span><br></pre></td></tr></table></figure>



<h4 id="4）slave1"><a href="#4）slave1" class="headerlink" title="4）slave1"></a>4）slave1</h4><p>服务器：192.168.25.138</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vi /usr/soft/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-a-s.properties</span><br></pre></td></tr></table></figure>



<p>修改配置如下：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#所属集群名字</span></span><br><span class="line">brokerClusterName=rocketmq-cluster</span><br><span class="line"><span class="comment">#broker名字，注意此处不同的配置文件填写的不一样</span></span><br><span class="line">brokerName=broker-a</span><br><span class="line"><span class="comment">#0 表示 Master，&gt;0 表示 Slave</span></span><br><span class="line">brokerId=1</span><br><span class="line"><span class="comment">#nameServer地址，分号分割</span></span><br><span class="line">namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876</span><br><span class="line"><span class="comment">#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数</span></span><br><span class="line">defaultTopicQueueNums=4</span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateTopicEnable=<span class="literal">true</span></span><br><span class="line"><span class="comment">#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭</span></span><br><span class="line">autoCreateSubscriptionGroup=<span class="literal">true</span></span><br><span class="line"><span class="comment">#Broker 对外服务的监听端口</span></span><br><span class="line">listenPort=11011</span><br><span class="line"><span class="comment">#删除文件时间点，默认凌晨 4点</span></span><br><span class="line">deleteWhen=04</span><br><span class="line"><span class="comment">#文件保留时间，默认 48 小时</span></span><br><span class="line">fileReservedTime=120</span><br><span class="line"><span class="comment">#commitLog每个文件的大小默认1G</span></span><br><span class="line">mapedFileSizeCommitLog=1073741824</span><br><span class="line"><span class="comment">#ConsumeQueue每个文件默认存30W条，根据业务情况调整</span></span><br><span class="line">mapedFileSizeConsumeQueue=300000</span><br><span class="line"><span class="comment">#destroyMapedFileIntervalForcibly=120000</span></span><br><span class="line"><span class="comment">#redeleteHangedFileInterval=120000</span></span><br><span class="line"><span class="comment">#检测物理文件磁盘空间</span></span><br><span class="line">diskMaxUsedSpaceRatio=88</span><br><span class="line"><span class="comment">#存储路径</span></span><br><span class="line">storePathRootDir=/usr/local/rocketmq/store</span><br><span class="line"><span class="comment">#commitLog 存储路径</span></span><br><span class="line">storePathCommitLog=/usr/local/rocketmq/store/commitlog</span><br><span class="line"><span class="comment">#消费队列存储路径存储路径</span></span><br><span class="line">storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue</span><br><span class="line"><span class="comment">#消息索引存储路径</span></span><br><span class="line">storePathIndex=/usr/local/rocketmq/store/index</span><br><span class="line"><span class="comment">#checkpoint 文件存储路径</span></span><br><span class="line">storeCheckpoint=/usr/local/rocketmq/store/checkpoint</span><br><span class="line"><span class="comment">#abort 文件存储路径</span></span><br><span class="line">abortFile=/usr/local/rocketmq/store/abort</span><br><span class="line"><span class="comment">#限制的消息大小</span></span><br><span class="line">maxMessageSize=65536</span><br><span class="line"><span class="comment">#flushCommitLogLeastPages=4</span></span><br><span class="line"><span class="comment">#flushConsumeQueueLeastPages=2</span></span><br><span class="line"><span class="comment">#flushCommitLogThoroughInterval=10000</span></span><br><span class="line"><span class="comment">#flushConsumeQueueThoroughInterval=60000</span></span><br><span class="line"><span class="comment">#Broker 的角色</span></span><br><span class="line"><span class="comment">#- ASYNC_MASTER 异步复制Master</span></span><br><span class="line"><span class="comment">#- SYNC_MASTER 同步双写Master</span></span><br><span class="line"><span class="comment">#- SLAVE</span></span><br><span class="line">brokerRole=SLAVE</span><br><span class="line"><span class="comment">#刷盘方式</span></span><br><span class="line"><span class="comment">#- ASYNC_FLUSH 异步刷盘</span></span><br><span class="line"><span class="comment">#- SYNC_FLUSH 同步刷盘</span></span><br><span class="line">flushDiskType=ASYNC_FLUSH</span><br><span class="line"><span class="comment">#checkTransactionMessageEnable=false</span></span><br><span class="line"><span class="comment">#发消息线程池数量</span></span><br><span class="line"><span class="comment">#sendMessageThreadPoolNums=128</span></span><br><span class="line"><span class="comment">#拉消息线程池数量</span></span><br><span class="line"><span class="comment">#pullMessageThreadPoolNums=128</span></span><br></pre></td></tr></table></figure>



<h3 id="3-3-9-修改启动脚本文件"><a href="#3-3-9-修改启动脚本文件" class="headerlink" title="3.3.9 修改启动脚本文件"></a>3.3.9 修改启动脚本文件</h3><h4 id="1）runbroker-sh"><a href="#1）runbroker-sh" class="headerlink" title="1）runbroker.sh"></a>1）runbroker.sh</h4><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vi /usr/local/rocketmq/bin/runbroker.sh</span><br></pre></td></tr></table></figure>



<p>需要根据内存大小进行适当的对JVM参数进行调整：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#===================================================</span></span><br><span class="line"><span class="comment"># 开发环境配置 JVM Configuration</span></span><br><span class="line">JAVA_OPT=<span class="string">&quot;<span class="variable">$&#123;JAVA_OPT&#125;</span> -server -Xms256m -Xmx256m -Xmn128m&quot;</span></span><br></pre></td></tr></table></figure>

<p>####2）runserver.sh</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim /usr/local/rocketmq/bin/runserver.sh</span><br></pre></td></tr></table></figure>



<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">JAVA_OPT=<span class="string">&quot;<span class="variable">$&#123;JAVA_OPT&#125;</span> -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m&quot;</span></span><br></pre></td></tr></table></figure>



<h3 id="3-3-10-服务启动"><a href="#3-3-10-服务启动" class="headerlink" title="3.3.10 服务启动"></a>3.3.10 服务启动</h3><h4 id="1）启动NameServe集群"><a href="#1）启动NameServe集群" class="headerlink" title="1）启动NameServe集群"></a>1）启动NameServe集群</h4><p>分别在192.168.25.135和192.168.25.138启动NameServer</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/local/rocketmq/bin</span><br><span class="line"><span class="built_in">nohup</span> sh mqnamesrv &amp;</span><br></pre></td></tr></table></figure>



<h4 id="2）启动Broker集群"><a href="#2）启动Broker集群" class="headerlink" title="2）启动Broker集群"></a>2）启动Broker集群</h4><ul>
<li>在192.168.25.135上启动master1和slave2</li>
</ul>
<p>master1：</p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/local/rocketmq/bin</span><br><span class="line"><span class="built_in">nohup</span> sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-syncbroker-a.properties &amp;</span><br></pre></td></tr></table></figure>



<p>slave2：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/local/rocketmq/bin</span><br><span class="line"><span class="built_in">nohup</span> sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-b-s.properties &amp;</span><br></pre></td></tr></table></figure>



<ul>
<li>在192.168.25.138上启动master2和slave2</li>
</ul>
<p>master2</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/local/rocketmq/bin</span><br><span class="line"><span class="built_in">nohup</span> sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-b.properties &amp;</span><br></pre></td></tr></table></figure>



<p>slave1</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/local/rocketmq/bin</span><br><span class="line"><span class="built_in">nohup</span> sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-<span class="built_in">sync</span>/broker-a-s.properties &amp;</span><br></pre></td></tr></table></figure>



<h3 id="3-3-11-查看进程状态"><a href="#3-3-11-查看进程状态" class="headerlink" title="3.3.11 查看进程状态"></a>3.3.11 查看进程状态</h3><p>启动后通过JPS查看启动进程</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/jps1.png"
                      alt="img"
                ></p>
<h3 id="3-3-12-查看日志"><a href="#3-3-12-查看日志" class="headerlink" title="3.3.12 查看日志"></a>3.3.12 查看日志</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># 查看nameServer日志</span></span><br><span class="line"><span class="built_in">tail</span> -500f ~/logs/rocketmqlogs/namesrv.log</span><br><span class="line"><span class="comment"># 查看broker日志</span></span><br><span class="line"><span class="built_in">tail</span> -500f ~/logs/rocketmqlogs/broker.log</span><br></pre></td></tr></table></figure>

<h2 id="3-4-mqadmin管理工具"><a href="#3-4-mqadmin管理工具" class="headerlink" title="3.4 mqadmin管理工具"></a>3.4 mqadmin管理工具</h2><h3 id="3-4-1-使用方式"><a href="#3-4-1-使用方式" class="headerlink" title="3.4.1 使用方式"></a>3.4.1 使用方式</h3><p>进入RocketMQ安装位置，在bin目录下执行<code>./mqadmin &#123;command&#125; &#123;args&#125;</code></p>
<p>###3.4.2 命令介绍</p>
<p>####1）Topic相关</p>
<table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>updateTopic</td>
<td>创建更新Topic配置</td>
<td>-b</td>
<td>Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port</td>
</tr>
<tr>
<td>-c</td>
<td>cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询）</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h-</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-p</td>
<td>指定新topic的读写权限( W&#x3D;2|R&#x3D;4|WR&#x3D;6 )</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-r</td>
<td>可读队列数（默认为 8）</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-w</td>
<td>可写队列数（默认为 8）</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）</td>
<td></td>
<td></td>
</tr>
<tr>
<td>deleteTopic</td>
<td>删除Topic</td>
<td>-c</td>
<td>cluster 名称，表示删除某集群下的某个 topic （集群 可通过 clusterList 查询）</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）</td>
<td></td>
<td></td>
</tr>
<tr>
<td>topicList</td>
<td>查看 Topic 列表信息</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-c</td>
<td>不配置-c只返回topic列表，增加-c返回clusterName, topic, consumerGroup信息，即topic的所属集群和订阅关系，没有参数</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>topicRoute</td>
<td>查看 Topic 路由信息</td>
<td>-t</td>
<td>topic 名称</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>topicStatus</td>
<td>查看 Topic 消息队列offset</td>
<td>-t</td>
<td>topic 名称</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>topicClusterList</td>
<td>查看 Topic 所在集群列表</td>
<td>-t</td>
<td>topic 名称</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>updateTopicPerm</td>
<td>更新 Topic 读写权限</td>
<td>-t</td>
<td>topic 名称</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-p</td>
<td>指定新 topic 的读写权限( W&#x3D;2|R&#x3D;4|WR&#x3D;6 )</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询），-b优先，如果没有-b，则对集群中所有Broker执行命令</td>
<td></td>
<td></td>
</tr>
<tr>
<td>updateOrderConf</td>
<td>从NameServer上创建、删除、获取特定命名空间的kv配置，目前还未启用</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic，键</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-v</td>
<td>orderConf，值</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-m</td>
<td>method，可选get、put、delete</td>
<td></td>
<td></td>
</tr>
<tr>
<td>allocateMQ</td>
<td>以平均负载算法计算消费者列表负载消息队列的负载结果</td>
<td>-t</td>
<td>topic 名称</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>ipList，用逗号分隔，计算这些ip去负载Topic的消息队列</td>
<td></td>
<td></td>
</tr>
<tr>
<td>statsAll</td>
<td>打印Topic订阅关系、TPS、积累量、24h读写总量等信息</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-a</td>
<td>是否只打印活跃topic</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>指定topic</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<p>####2）集群相关</p>
<table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>clusterList</td>
<td>查看集群信息，集群、BrokerName、BrokerId、TPS等信息</td>
<td>-m</td>
<td>打印更多信息 (增加打印出如下信息 #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday)</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>打印间隔，单位秒</td>
<td></td>
<td></td>
</tr>
<tr>
<td>clusterRT</td>
<td>发送消息检测集群各Broker RT。消息发往${BrokerName} Topic。</td>
<td>-a</td>
<td>amount，每次探测的总数，RT &#x3D; 总时间 &#x2F; amount</td>
</tr>
<tr>
<td>-s</td>
<td>消息大小，单位B</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>探测哪个集群</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-p</td>
<td>是否打印格式化日志，以|分割，默认不打印</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-m</td>
<td>所属机房，打印使用</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>发送间隔，单位秒</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<p>####3）Broker相关</p>
<table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>updateBrokerConfig</td>
<td>更新 Broker 配置文件，会修改Broker.conf</td>
<td>-b</td>
<td>Broker 地址，格式为ip:port</td>
</tr>
<tr>
<td>-c</td>
<td>cluster 名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-k</td>
<td>key 值</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-v</td>
<td>value 值</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>brokerStatus</td>
<td>查看 Broker 统计信息、运行状态（你想要的信息几乎都在里面）</td>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>brokerConsumeStats</td>
<td>Broker中各个消费者的消费情况，按Message Queue维度返回Consume Offset，Broker Offset，Diff，TImestamp等信息</td>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
</tr>
<tr>
<td>-t</td>
<td>请求超时时间</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-l</td>
<td>diff阈值，超过阈值才打印</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-o</td>
<td>是否为顺序topic，一般为false</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>getBrokerConfig</td>
<td>获取Broker配置</td>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>wipeWritePerm</td>
<td>从NameServer上清除 Broker写权限</td>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>cleanExpiredCQ</td>
<td>清理Broker上过期的Consume Queue，如果手动减少对列数可能产生过期队列</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>集群名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>cleanUnusedTopic</td>
<td>清理Broker上不使用的Topic，从内存中释放Topic的Consume Queue，如果手动删除Topic会产生不使用的Topic</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>Broker 地址，地址为ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>集群名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>sendMsgStatus</td>
<td>向Broker发消息，返回发送状态和RT</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>BrokerName，注意不同于Broker地址</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>消息大小，单位B</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>发送次数</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<p>####4）消息相关</p>
<table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>queryMsgById</td>
<td>根据offsetMsgId查询msg，如果使用开源控制台，应使用offsetMsgId，此命令还有其他参数，具体作用请阅读QueryMsgByIdSubCommand。</td>
<td>-i</td>
<td>msgId</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>queryMsgByKey</td>
<td>根据消息 Key 查询消息</td>
<td>-k</td>
<td>msgKey</td>
</tr>
<tr>
<td>-t</td>
<td>Topic 名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>queryMsgByOffset</td>
<td>根据 Offset 查询消息</td>
<td>-b</td>
<td>Broker 名称，（这里需要注意 填写的是 Broker 的名称，不是 Broker 的地址，Broker 名称可以在 clusterList 查到）</td>
</tr>
<tr>
<td>-i</td>
<td>query 队列 id</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-o</td>
<td>offset 值</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic 名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>queryMsgByUniqueKey</td>
<td>根据msgId查询，msgId不同于offsetMsgId，区别详见常见运维问题。-g，-d配合使用，查到消息后尝试让特定的消费者消费消息并返回消费结果</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>uniqe msg id</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>consumerGroup</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>clientId</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>checkMsgSendRT</td>
<td>检测向topic发消息的RT，功能类似clusterRT</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-a</td>
<td>探测次数</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>消息大小</td>
<td></td>
<td></td>
</tr>
<tr>
<td>sendMessage</td>
<td>发送一条消息，可以根据配置发往特定Message Queue，或普通发送。</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-p</td>
<td>body，消息体</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-k</td>
<td>keys</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>tags</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>BrokerName</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>queueId</td>
<td></td>
<td></td>
</tr>
<tr>
<td>consumeMessage</td>
<td>消费消息。可以根据offset、开始&amp;结束时间戳、消息队列消费消息，配置不同执行不同消费逻辑，详见ConsumeMessageCommand。</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>BrokerName</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-o</td>
<td>从offset开始消费</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>queueId</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>消费者分组</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>开始时间戳，格式详见-h</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>结束时间戳</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>消费多少条消息</td>
<td></td>
<td></td>
</tr>
<tr>
<td>printMsg</td>
<td>从Broker消费消息并打印，可选时间段</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>字符集，例如UTF-8</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>subExpress，过滤表达式</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>开始时间戳，格式参见-h</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-e</td>
<td>结束时间戳</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>是否打印消息体</td>
<td></td>
<td></td>
</tr>
<tr>
<td>printMsgByQueue</td>
<td>类似printMsg，但指定Message Queue</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>queueId</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-a</td>
<td>BrokerName</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>字符集，例如UTF-8</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>subExpress，过滤表达式</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>开始时间戳，格式参见-h</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-e</td>
<td>结束时间戳</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-p</td>
<td>是否打印消息</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>是否打印消息体</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-f</td>
<td>是否统计tag数量并打印</td>
<td></td>
<td></td>
</tr>
<tr>
<td>resetOffsetByTime</td>
<td>按时间戳重置offset，Broker和consumer都会重置</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>消费者分组</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>重置为此时间戳对应的offset</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-f</td>
<td>是否强制重置，如果false，只支持回溯offset，如果true，不管时间戳对应offset与consumeOffset关系</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>是否重置c++客户端offset</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<h4 id="5）消费者、消费组相关"><a href="#5）消费者、消费组相关" class="headerlink" title="#5）消费者、消费组相关"></a><a class="link"   target="_blank" rel="noopener" href="https://pchaoo.gitee.io/blog/views/rocketmq/RocketMQ-01.html#_5%EF%BC%89%E6%B6%88%E8%B4%B9%E8%80%85%E3%80%81%E6%B6%88%E8%B4%B9%E7%BB%84%E7%9B%B8%E5%85%B3" >#<i class="fas fa-external-link-alt"></i></a>5）消费者、消费组相关</h4><table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>consumerProgress</td>
<td>查看订阅组消费状态，可以查看具体的client IP的消息积累量</td>
<td>-g</td>
<td>消费者所属组名</td>
</tr>
<tr>
<td>-s</td>
<td>是否打印client IP</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>consumerStatus</td>
<td>查看消费者状态，包括同一个分组中是否都是相同的订阅，分析Process Queue是否堆积，返回消费者jstack结果，内容较多，使用者参见ConsumerStatusSubCommand</td>
<td>-h</td>
<td>打印帮助</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>consumer group</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>clientId</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>是否执行jstack</td>
<td></td>
<td></td>
</tr>
<tr>
<td>getConsumerStatus</td>
<td>获取 Consumer 消费进度</td>
<td>-g</td>
<td>消费者所属组名</td>
</tr>
<tr>
<td>-t</td>
<td>查询主题</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>Consumer 客户端 ip</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>updateSubGroup</td>
<td>更新或创建订阅关系</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>Broker地址</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>集群名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>消费者分组名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>分组是否允许消费</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-m</td>
<td>是否从最小offset开始消费</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>是否是广播模式</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-q</td>
<td>重试队列数量</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-r</td>
<td>最大重试次数</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-i</td>
<td>当slaveReadEnable开启时有效，且还未达到从slave消费时建议从哪个BrokerId消费，可以配置备机id，主动从备机消费</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-w</td>
<td>如果Broker建议从slave消费，配置决定从哪个slave消费，配置BrokerId，例如1</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-a</td>
<td>当消费者数量变化时是否通知其他消费者负载均衡</td>
<td></td>
<td></td>
</tr>
<tr>
<td>deleteSubGroup</td>
<td>从Broker删除订阅关系</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-b</td>
<td>Broker地址</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-c</td>
<td>集群名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-g</td>
<td>消费者分组名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>cloneGroupOffset</td>
<td>在目标群组中使用源群组的offset</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-s</td>
<td>源消费者组</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-d</td>
<td>目标消费者组</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-t</td>
<td>topic名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-o</td>
<td>暂未使用</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<h4 id="6）连接相关"><a href="#6）连接相关" class="headerlink" title="6）连接相关"></a>6）连接相关</h4><table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>consumerConnec tion</td>
<td>查询 Consumer 的网络连接</td>
<td>-g</td>
<td>消费者所属组名</td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>producerConnec tion</td>
<td>查询 Producer 的网络连接</td>
<td>-g</td>
<td>生产者所属组名</td>
</tr>
<tr>
<td>-t</td>
<td>主题名称</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<h4 id="7）NameServer相关"><a href="#7）NameServer相关" class="headerlink" title="7）NameServer相关"></a>7）NameServer相关</h4><table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>updateKvConfig</td>
<td>更新NameServer的kv配置，目前还未使用</td>
<td>-s</td>
<td>命名空间</td>
</tr>
<tr>
<td>-k</td>
<td>key</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-v</td>
<td>value</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>deleteKvConfig</td>
<td>删除NameServer的kv配置</td>
<td>-s</td>
<td>命名空间</td>
</tr>
<tr>
<td>-k</td>
<td>key</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>getNamesrvConfig</td>
<td>获取NameServer配置</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>updateNamesrvConfig</td>
<td>修改NameServer配置</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-k</td>
<td>key</td>
<td></td>
<td></td>
</tr>
<tr>
<td>-v</td>
<td>value</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<h4 id="8）其他"><a href="#8）其他" class="headerlink" title="8）其他"></a>8）其他</h4><table>
<thead>
<tr>
<th>名称</th>
<th>含义</th>
<th>命令选项</th>
<th>说明</th>
</tr>
</thead>
<tbody><tr>
<td>startMonitoring</td>
<td>开启监控进程，监控消息误删、重试队列消息数等</td>
<td>-n</td>
<td>NameServer 服务地址，格式 ip:port</td>
</tr>
<tr>
<td>-h</td>
<td>打印帮助</td>
<td></td>
<td></td>
</tr>
</tbody></table>
<h3 id="3-4-3-注意事项"><a href="#3-4-3-注意事项" class="headerlink" title="3.4.3 注意事项"></a>3.4.3 注意事项</h3><ul>
<li>几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port</li>
<li>几乎所有命令都可以通过-h获取帮助</li>
<li>如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令；如果不配置Broker地址，则对集群中所有主机执行命令</li>
</ul>
<h2 id="3-5-集群监控平台搭建"><a href="#3-5-集群监控平台搭建" class="headerlink" title="3.5 集群监控平台搭建"></a>3.5 集群监控平台搭建</h2><h3 id="3-5-1-概述"><a href="#3-5-1-概述" class="headerlink" title="3.5.1 概述"></a>3.5.1 概述</h3><p><code>RocketMQ</code>有一个对其扩展的开源项目<a class="link"   target="_blank" rel="noopener" href="https://github.com/apache/rocketmq-externals" >incubator-rocketmq-externals<i class="fas fa-external-link-alt"></i></a>，这个项目中有一个子模块叫<code>rocketmq-console</code>，这个便是管理控制台项目了，先将<a class="link"   target="_blank" rel="noopener" href="https://github.com/apache/rocketmq-externals" >incubator-rocketmq-externals<i class="fas fa-external-link-alt"></i></a>拉到本地，因为我们需要自己对<code>rocketmq-console</code>进行编译打包运行。</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/rocketmq-console.png"
                      alt="img"
                ></p>
<h3 id="3-5-2-下载并编译打包"><a href="#3-5-2-下载并编译打包" class="headerlink" title="3.5.2 下载并编译打包"></a>3.5.2 下载并编译打包</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">git <span class="built_in">clone</span> https://github.com/apache/rocketmq-externals</span><br><span class="line"><span class="built_in">cd</span> rocketmq-console</span><br><span class="line">mvn clean package -Dmaven.test.skip=<span class="literal">true</span></span><br></pre></td></tr></table></figure>



<p>注意：打包前在<code>rocketmq-console</code>中配置<code>namesrv</code>集群地址：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876</span><br></pre></td></tr></table></figure>



<p>启动rocketmq-console：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">java -jar rocketmq-console-ng-1.0.0.jar</span><br></pre></td></tr></table></figure>



<p>启动成功后，我们就可以通过浏览器访问<code>http://localhost:8080</code>进入控制台界面了，如下图：</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/rocketmq-console2.png"
                      alt="img"
                ></p>
<p>集群状态：</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/rocketmq-console3.png"
                      alt="img"
                ></p>
<h1 id="4-消息发送样例"><a href="#4-消息发送样例" class="headerlink" title="4. 消息发送样例"></a>4. 消息发送样例</h1><ul>
<li>导入MQ客户端依赖</li>
</ul>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.rocketmq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>rocketmq-client<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>4.4.0<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>



<ul>
<li>消息发送者步骤分析r</li>
</ul>
<figure class="highlight tex"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line">1.创建消息生产者producer，并制定生产者组名</span><br><span class="line">2.指定Nameserver地址</span><br><span class="line">3.启动producer</span><br><span class="line">4.创建消息对象，指定主题Topic、Tag和消息体</span><br><span class="line">5.发送消息</span><br><span class="line">6.关闭生产者producer</span><br></pre></td></tr></table></figure>



<ul>
<li>消息消费者步骤分析</li>
</ul>
<figure class="highlight tex"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">1.创建消费者Consumer，制定消费者组名</span><br><span class="line">2.指定Nameserver地址</span><br><span class="line">3.订阅主题Topic和Tag</span><br><span class="line">4.设置回调函数，处理消息</span><br><span class="line">5.启动消费者consumer</span><br></pre></td></tr></table></figure>



<h2 id="4-1-基本样例"><a href="#4-1-基本样例" class="headerlink" title="4.1 基本样例"></a>4.1 基本样例</h2><h3 id="4-1-1-消息发送"><a href="#4-1-1-消息发送" class="headerlink" title="4.1.1 消息发送"></a>4.1.1 消息发送</h3><h4 id="1）发送同步消息"><a href="#1）发送同步消息" class="headerlink" title="1）发送同步消息"></a>1）发送同步消息</h4><p>这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">SyncProducer</span> &#123;</span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        <span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;please_rename_unique_group_name&quot;</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">    	producer.setNamesrvAddr(<span class="string">&quot;localhost:9876&quot;</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">    	<span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">    	    <span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">    	    <span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TopicTest&quot;</span> <span class="comment">/* Topic */</span>,</span><br><span class="line">        	<span class="string">&quot;TagA&quot;</span> <span class="comment">/* Tag */</span>,</span><br><span class="line">        	(<span class="string">&quot;Hello RocketMQ &quot;</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET) <span class="comment">/* Message body */</span></span><br><span class="line">        	);</span><br><span class="line">        	<span class="comment">// 发送消息到一个Broker</span></span><br><span class="line">            <span class="type">SendResult</span> <span class="variable">sendResult</span> <span class="operator">=</span> producer.send(msg);</span><br><span class="line">            <span class="comment">// 通过sendResult返回消息是否成功送达</span></span><br><span class="line">            System.out.printf(<span class="string">&quot;%s%n&quot;</span>, sendResult);</span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h4 id="2）发送异步消息"><a href="#2）发送异步消息" class="headerlink" title="2）发送异步消息"></a>2）发送异步消息</h4><p>异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">AsyncProducer</span> &#123;</span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        <span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;please_rename_unique_group_name&quot;</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">        producer.setNamesrvAddr(<span class="string">&quot;localhost:9876&quot;</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">        producer.setRetryTimesWhenSendAsyncFailed(<span class="number">0</span>);</span><br><span class="line">    	<span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">                <span class="keyword">final</span> <span class="type">int</span> <span class="variable">index</span> <span class="operator">=</span> i;</span><br><span class="line">            	<span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">                <span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TopicTest&quot;</span>,</span><br><span class="line">                    <span class="string">&quot;TagA&quot;</span>,</span><br><span class="line">                    <span class="string">&quot;OrderID188&quot;</span>,</span><br><span class="line">                    <span class="string">&quot;Hello world&quot;</span>.getBytes(RemotingHelper.DEFAULT_CHARSET));</span><br><span class="line">                <span class="comment">// SendCallback接收异步返回结果的回调</span></span><br><span class="line">                producer.send(msg, <span class="keyword">new</span> <span class="title class_">SendCallback</span>() &#123;</span><br><span class="line">                    <span class="meta">@Override</span></span><br><span class="line">                    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">onSuccess</span><span class="params">(SendResult sendResult)</span> &#123;</span><br><span class="line">                        System.out.printf(<span class="string">&quot;%-10d OK %s %n&quot;</span>, index,</span><br><span class="line">                            sendResult.getMsgId());</span><br><span class="line">                    &#125;</span><br><span class="line">                    <span class="meta">@Override</span></span><br><span class="line">                    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">onException</span><span class="params">(Throwable e)</span> &#123;</span><br><span class="line">      	              System.out.printf(<span class="string">&quot;%-10d Exception %s %n&quot;</span>, index, e);</span><br><span class="line">      	              e.printStackTrace();</span><br><span class="line">                    &#125;</span><br><span class="line">            	&#125;);</span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h4 id="3）单向发送消息"><a href="#3）单向发送消息" class="headerlink" title="3）单向发送消息"></a>3）单向发送消息</h4><p>这种方式主要用在不特别关心发送结果的场景，例如日志发送。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">OnewayProducer</span> &#123;</span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception&#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        <span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;please_rename_unique_group_name&quot;</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">        producer.setNamesrvAddr(<span class="string">&quot;localhost:9876&quot;</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">    	<span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">        	<span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">        	<span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TopicTest&quot;</span> <span class="comment">/* Topic */</span>,</span><br><span class="line">                <span class="string">&quot;TagA&quot;</span> <span class="comment">/* Tag */</span>,</span><br><span class="line">                (<span class="string">&quot;Hello RocketMQ &quot;</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET) <span class="comment">/* Message body */</span></span><br><span class="line">        	);</span><br><span class="line">        	<span class="comment">// 发送单向消息，没有任何返回结果</span></span><br><span class="line">        	producer.sendOneway(msg);</span><br><span class="line"></span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h3 id="4-1-2-消费消息"><a href="#4-1-2-消费消息" class="headerlink" title="4.1.2 消费消息"></a>4.1.2 消费消息</h3><h4 id="1）负载均衡模式"><a href="#1）负载均衡模式" class="headerlink" title="1）负载均衡模式"></a>1）负载均衡模式</h4><p>消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">    <span class="comment">// 实例化消息生产者,指定组名</span></span><br><span class="line">    <span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;group1&quot;</span>);</span><br><span class="line">    <span class="comment">// 指定Namesrv地址信息.</span></span><br><span class="line">    consumer.setNamesrvAddr(<span class="string">&quot;localhost:9876&quot;</span>);</span><br><span class="line">    <span class="comment">// 订阅Topic</span></span><br><span class="line">    consumer.subscribe(<span class="string">&quot;Test&quot;</span>, <span class="string">&quot;*&quot;</span>);</span><br><span class="line">    <span class="comment">//负载均衡模式消费</span></span><br><span class="line">    consumer.setMessageModel(MessageModel.CLUSTERING);</span><br><span class="line">    <span class="comment">// 注册回调函数，处理消息</span></span><br><span class="line">    consumer.registerMessageListener(<span class="keyword">new</span> <span class="title class_">MessageListenerConcurrently</span>() &#123;</span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="keyword">public</span> ConsumeConcurrentlyStatus <span class="title function_">consumeMessage</span><span class="params">(List&lt;MessageExt&gt; msgs,</span></span><br><span class="line"><span class="params">                                                        ConsumeConcurrentlyContext context)</span> &#123;</span><br><span class="line">            System.out.printf(<span class="string">&quot;%s Receive New Messages: %s %n&quot;</span>, </span><br><span class="line">                              Thread.currentThread().getName(), msgs);</span><br><span class="line">            <span class="keyword">return</span> ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;);</span><br><span class="line">    <span class="comment">//启动消息者</span></span><br><span class="line">    consumer.start();</span><br><span class="line">    System.out.printf(<span class="string">&quot;Consumer Started.%n&quot;</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h4 id="2）广播模式"><a href="#2）广播模式" class="headerlink" title="2）广播模式"></a>2）广播模式</h4><p>消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">    <span class="comment">// 实例化消息生产者,指定组名</span></span><br><span class="line">    <span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;group1&quot;</span>);</span><br><span class="line">    <span class="comment">// 指定Namesrv地址信息.</span></span><br><span class="line">    consumer.setNamesrvAddr(<span class="string">&quot;localhost:9876&quot;</span>);</span><br><span class="line">    <span class="comment">// 订阅Topic</span></span><br><span class="line">    consumer.subscribe(<span class="string">&quot;Test&quot;</span>, <span class="string">&quot;*&quot;</span>);</span><br><span class="line">    <span class="comment">//广播模式消费</span></span><br><span class="line">    consumer.setMessageModel(MessageModel.BROADCASTING);</span><br><span class="line">    <span class="comment">// 注册回调函数，处理消息</span></span><br><span class="line">    consumer.registerMessageListener(<span class="keyword">new</span> <span class="title class_">MessageListenerConcurrently</span>() &#123;</span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="keyword">public</span> ConsumeConcurrentlyStatus <span class="title function_">consumeMessage</span><span class="params">(List&lt;MessageExt&gt; msgs,</span></span><br><span class="line"><span class="params">                                                        ConsumeConcurrentlyContext context)</span> &#123;</span><br><span class="line">            System.out.printf(<span class="string">&quot;%s Receive New Messages: %s %n&quot;</span>, </span><br><span class="line">                              Thread.currentThread().getName(), msgs);</span><br><span class="line">            <span class="keyword">return</span> ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;);</span><br><span class="line">    <span class="comment">//启动消息者</span></span><br><span class="line">    consumer.start();</span><br><span class="line">    System.out.printf(<span class="string">&quot;Consumer Started.%n&quot;</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h2 id="4-2-顺序消息"><a href="#4-2-顺序消息" class="headerlink" title="4.2 顺序消息"></a>4.2 顺序消息</h2><p>消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。</p>
<p>顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。</p>
<p>下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。</p>
<h3 id="4-2-1-顺序消息生产"><a href="#4-2-1-顺序消息生产" class="headerlink" title="4.2.1 顺序消息生产"></a>4.2.1 顺序消息生产</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">* Producer，发送顺序消息</span></span><br><span class="line"><span class="comment">*/</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">Producer</span> &#123;</span><br><span class="line"></span><br><span class="line">   <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">       <span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;please_rename_unique_group_name&quot;</span>);</span><br><span class="line"></span><br><span class="line">       producer.setNamesrvAddr(<span class="string">&quot;127.0.0.1:9876&quot;</span>);</span><br><span class="line"></span><br><span class="line">       producer.start();</span><br><span class="line"></span><br><span class="line">       String[] tags = <span class="keyword">new</span> <span class="title class_">String</span>[]&#123;<span class="string">&quot;TagA&quot;</span>, <span class="string">&quot;TagC&quot;</span>, <span class="string">&quot;TagD&quot;</span>&#125;;</span><br><span class="line"></span><br><span class="line">       <span class="comment">// 订单列表</span></span><br><span class="line">       List&lt;OrderStep&gt; orderList = <span class="keyword">new</span> <span class="title class_">Producer</span>().buildOrders();</span><br><span class="line"></span><br><span class="line">       <span class="type">Date</span> <span class="variable">date</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Date</span>();</span><br><span class="line">       <span class="type">SimpleDateFormat</span> <span class="variable">sdf</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">SimpleDateFormat</span>(<span class="string">&quot;yyyy-MM-dd HH:mm:ss&quot;</span>);</span><br><span class="line">       <span class="type">String</span> <span class="variable">dateStr</span> <span class="operator">=</span> sdf.format(date);</span><br><span class="line">       <span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; <span class="number">10</span>; i++) &#123;</span><br><span class="line">           <span class="comment">// 加个时间前缀</span></span><br><span class="line">           <span class="type">String</span> <span class="variable">body</span> <span class="operator">=</span> dateStr + <span class="string">&quot; Hello RocketMQ &quot;</span> + orderList.get(i);</span><br><span class="line">           <span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TopicTest&quot;</span>, tags[i % tags.length], <span class="string">&quot;KEY&quot;</span> + i, body.getBytes());</span><br><span class="line"></span><br><span class="line">           <span class="type">SendResult</span> <span class="variable">sendResult</span> <span class="operator">=</span> producer.send(msg, <span class="keyword">new</span> <span class="title class_">MessageQueueSelector</span>() &#123;</span><br><span class="line">               <span class="meta">@Override</span></span><br><span class="line">               <span class="keyword">public</span> MessageQueue <span class="title function_">select</span><span class="params">(List&lt;MessageQueue&gt; mqs, Message msg, Object arg)</span> &#123;</span><br><span class="line">                   <span class="type">Long</span> <span class="variable">id</span> <span class="operator">=</span> (Long) arg;  <span class="comment">//根据订单id选择发送queue</span></span><br><span class="line">                   <span class="type">long</span> <span class="variable">index</span> <span class="operator">=</span> id % mqs.size();</span><br><span class="line">                   <span class="keyword">return</span> mqs.get((<span class="type">int</span>) index);</span><br><span class="line">               &#125;</span><br><span class="line">           &#125;, orderList.get(i).getOrderId());<span class="comment">//订单id</span></span><br><span class="line"></span><br><span class="line">           System.out.println(String.format(<span class="string">&quot;SendResult status:%s, queueId:%d, body:%s&quot;</span>,</span><br><span class="line">               sendResult.getSendStatus(),</span><br><span class="line">               sendResult.getMessageQueue().getQueueId(),</span><br><span class="line">               body));</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       producer.shutdown();</span><br><span class="line">   &#125;</span><br><span class="line"></span><br><span class="line">   <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 订单的步骤</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">class</span> <span class="title class_">OrderStep</span> &#123;</span><br><span class="line">       <span class="keyword">private</span> <span class="type">long</span> orderId;</span><br><span class="line">       <span class="keyword">private</span> String desc;</span><br><span class="line"></span><br><span class="line">       <span class="keyword">public</span> <span class="type">long</span> <span class="title function_">getOrderId</span><span class="params">()</span> &#123;</span><br><span class="line">           <span class="keyword">return</span> orderId;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">setOrderId</span><span class="params">(<span class="type">long</span> orderId)</span> &#123;</span><br><span class="line">           <span class="built_in">this</span>.orderId = orderId;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       <span class="keyword">public</span> String <span class="title function_">getDesc</span><span class="params">()</span> &#123;</span><br><span class="line">           <span class="keyword">return</span> desc;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">setDesc</span><span class="params">(String desc)</span> &#123;</span><br><span class="line">           <span class="built_in">this</span>.desc = desc;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       <span class="meta">@Override</span></span><br><span class="line">       <span class="keyword">public</span> String <span class="title function_">toString</span><span class="params">()</span> &#123;</span><br><span class="line">           <span class="keyword">return</span> <span class="string">&quot;OrderStep&#123;&quot;</span> +</span><br><span class="line">               <span class="string">&quot;orderId=&quot;</span> + orderId +</span><br><span class="line">               <span class="string">&quot;, desc=&#x27;&quot;</span> + desc + <span class="string">&#x27;\&#x27;&#x27;</span> +</span><br><span class="line">               <span class="string">&#x27;&#125;&#x27;</span>;</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;</span><br><span class="line"></span><br><span class="line">   <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 生成模拟订单数据</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="keyword">private</span> List&lt;OrderStep&gt; <span class="title function_">buildOrders</span><span class="params">()</span> &#123;</span><br><span class="line">       List&lt;OrderStep&gt; orderList = <span class="keyword">new</span> <span class="title class_">ArrayList</span>&lt;OrderStep&gt;();</span><br><span class="line"></span><br><span class="line">       <span class="type">OrderStep</span> <span class="variable">orderDemo</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;创建&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;创建&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;付款&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;创建&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;付款&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;付款&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;完成&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;推送&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;完成&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       orderDemo = <span class="keyword">new</span> <span class="title class_">OrderStep</span>();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">&quot;完成&quot;</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"></span><br><span class="line">       <span class="keyword">return</span> orderList;</span><br><span class="line">   &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h3 id="4-2-2-顺序消费消息"><a href="#4-2-2-顺序消费消息" class="headerlink" title="4.2.2 顺序消费消息"></a>4.2.2 顺序消费消息</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）</span></span><br><span class="line"><span class="comment">*/</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">ConsumerInOrder</span> &#123;</span><br><span class="line"></span><br><span class="line">   <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">       <span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> </span><br><span class="line">           <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;please_rename_unique_group_name_3&quot;</span>);</span><br><span class="line">       consumer.setNamesrvAddr(<span class="string">&quot;127.0.0.1:9876&quot;</span>);</span><br><span class="line">       <span class="comment">/**</span></span><br><span class="line"><span class="comment">        * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费&lt;br&gt;</span></span><br><span class="line"><span class="comment">        * 如果非第一次启动，那么按照上次消费的位置继续消费</span></span><br><span class="line"><span class="comment">        */</span></span><br><span class="line">       consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);</span><br><span class="line"></span><br><span class="line">       consumer.subscribe(<span class="string">&quot;TopicTest&quot;</span>, <span class="string">&quot;TagA || TagC || TagD&quot;</span>);</span><br><span class="line"></span><br><span class="line">       consumer.registerMessageListener(<span class="keyword">new</span> <span class="title class_">MessageListenerOrderly</span>() &#123;</span><br><span class="line"></span><br><span class="line">           <span class="type">Random</span> <span class="variable">random</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Random</span>();</span><br><span class="line"></span><br><span class="line">           <span class="meta">@Override</span></span><br><span class="line">           <span class="keyword">public</span> ConsumeOrderlyStatus <span class="title function_">consumeMessage</span><span class="params">(List&lt;MessageExt&gt; msgs, ConsumeOrderlyContext context)</span> &#123;</span><br><span class="line">               context.setAutoCommit(<span class="literal">true</span>);</span><br><span class="line">               <span class="keyword">for</span> (MessageExt msg : msgs) &#123;</span><br><span class="line">                   <span class="comment">// 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序</span></span><br><span class="line">                   System.out.println(<span class="string">&quot;consumeThread=&quot;</span> + Thread.currentThread().getName() + <span class="string">&quot;queueId=&quot;</span> + msg.getQueueId() + <span class="string">&quot;, content:&quot;</span> + <span class="keyword">new</span> <span class="title class_">String</span>(msg.getBody()));</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               <span class="keyword">try</span> &#123;</span><br><span class="line">                   <span class="comment">//模拟业务逻辑处理中...</span></span><br><span class="line">                   TimeUnit.SECONDS.sleep(random.nextInt(<span class="number">10</span>));</span><br><span class="line">               &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">                   e.printStackTrace();</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="keyword">return</span> ConsumeOrderlyStatus.SUCCESS;</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;);</span><br><span class="line"></span><br><span class="line">       consumer.start();</span><br><span class="line"></span><br><span class="line">       System.out.println(<span class="string">&quot;Consumer Started.&quot;</span>);</span><br><span class="line">   &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h2 id="4-3-延时消息"><a href="#4-3-延时消息" class="headerlink" title="4.3 延时消息"></a>4.3 延时消息</h2><p>比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。</p>
<h3 id="4-3-1-启动消息消费者"><a href="#4-3-1-启动消息消费者" class="headerlink" title="4.3.1 启动消息消费者"></a>4.3.1 启动消息消费者</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">ScheduledMessageConsumer</span> &#123;</span><br><span class="line">   <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">      <span class="comment">// 实例化消费者</span></span><br><span class="line">      <span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;ExampleConsumer&quot;</span>);</span><br><span class="line">      <span class="comment">// 订阅Topics</span></span><br><span class="line">      consumer.subscribe(<span class="string">&quot;TestTopic&quot;</span>, <span class="string">&quot;*&quot;</span>);</span><br><span class="line">      <span class="comment">// 注册消息监听者</span></span><br><span class="line">      consumer.registerMessageListener(<span class="keyword">new</span> <span class="title class_">MessageListenerConcurrently</span>() &#123;</span><br><span class="line">          <span class="meta">@Override</span></span><br><span class="line">          <span class="keyword">public</span> ConsumeConcurrentlyStatus <span class="title function_">consumeMessage</span><span class="params">(List&lt;MessageExt&gt; messages, ConsumeConcurrentlyContext context)</span> &#123;</span><br><span class="line">              <span class="keyword">for</span> (MessageExt message : messages) &#123;</span><br><span class="line">                  <span class="comment">// Print approximate delay time period</span></span><br><span class="line">                  System.out.println(<span class="string">&quot;Receive message[msgId=&quot;</span> + message.getMsgId() + <span class="string">&quot;] &quot;</span> + (System.currentTimeMillis() - message.getStoreTimestamp()) + <span class="string">&quot;ms later&quot;</span>);</span><br><span class="line">              &#125;</span><br><span class="line">              <span class="keyword">return</span> ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</span><br><span class="line">          &#125;</span><br><span class="line">      &#125;);</span><br><span class="line">      <span class="comment">// 启动消费者</span></span><br><span class="line">      consumer.start();</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h3 id="4-3-2-发送延时消息"><a href="#4-3-2-发送延时消息" class="headerlink" title="4.3.2 发送延时消息"></a>4.3.2 发送延时消息</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">ScheduledMessageProducer</span> &#123;</span><br><span class="line">   <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception &#123;</span><br><span class="line">      <span class="comment">// 实例化一个生产者来产生延时消息</span></span><br><span class="line">      <span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;ExampleProducerGroup&quot;</span>);</span><br><span class="line">      <span class="comment">// 启动生产者</span></span><br><span class="line">      producer.start();</span><br><span class="line">      <span class="type">int</span> <span class="variable">totalMessagesToSend</span> <span class="operator">=</span> <span class="number">100</span>;</span><br><span class="line">      <span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; totalMessagesToSend; i++) &#123;</span><br><span class="line">          <span class="type">Message</span> <span class="variable">message</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TestTopic&quot;</span>, (<span class="string">&quot;Hello scheduled message &quot;</span> + i).getBytes());</span><br><span class="line">          <span class="comment">// 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)</span></span><br><span class="line">          message.setDelayTimeLevel(<span class="number">3</span>);</span><br><span class="line">          <span class="comment">// 发送消息</span></span><br><span class="line">          producer.send(message);</span><br><span class="line">      &#125;</span><br><span class="line">       <span class="comment">// 关闭生产者</span></span><br><span class="line">      producer.shutdown();</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<p>###4.3.3 验证</p>
<p>您将会看到消息的消费比存储时间晚10秒</p>
<h3 id="4-3-4-使用限制"><a href="#4-3-4-使用限制" class="headerlink" title="4.3.4 使用限制"></a>4.3.4 使用限制</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// org/apache/rocketmq/store/config/MessageStoreConfig.java</span></span><br><span class="line"><span class="keyword">private</span> <span class="type">String</span> <span class="variable">messageDelayLevel</span> <span class="operator">=</span> <span class="string">&quot;1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h&quot;</span>;</span><br></pre></td></tr></table></figure>



<p>现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18</p>
<h2 id="4-4-批量消息"><a href="#4-4-批量消息" class="headerlink" title="4.4 批量消息"></a>4.4 批量消息</h2><p>批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。</p>
<h3 id="4-4-1-发送批量消息"><a href="#4-4-1-发送批量消息" class="headerlink" title="4.4.1 发送批量消息"></a>4.4.1 发送批量消息</h3><p>如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="type">String</span> <span class="variable">topic</span> <span class="operator">=</span> <span class="string">&quot;BatchTest&quot;</span>;</span><br><span class="line">List&lt;Message&gt; messages = <span class="keyword">new</span> <span class="title class_">ArrayList</span>&lt;&gt;();</span><br><span class="line">messages.add(<span class="keyword">new</span> <span class="title class_">Message</span>(topic, <span class="string">&quot;TagA&quot;</span>, <span class="string">&quot;OrderID001&quot;</span>, <span class="string">&quot;Hello world 0&quot;</span>.getBytes()));</span><br><span class="line">messages.add(<span class="keyword">new</span> <span class="title class_">Message</span>(topic, <span class="string">&quot;TagA&quot;</span>, <span class="string">&quot;OrderID002&quot;</span>, <span class="string">&quot;Hello world 1&quot;</span>.getBytes()));</span><br><span class="line">messages.add(<span class="keyword">new</span> <span class="title class_">Message</span>(topic, <span class="string">&quot;TagA&quot;</span>, <span class="string">&quot;OrderID003&quot;</span>, <span class="string">&quot;Hello world 2&quot;</span>.getBytes()));</span><br><span class="line"><span class="keyword">try</span> &#123;</span><br><span class="line">   producer.send(messages);</span><br><span class="line">&#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">   e.printStackTrace();</span><br><span class="line">   <span class="comment">//处理error</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<p>如果消息的总长度可能大于4MB时，这时候最好把消息进行分割</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">ListSplitter</span> <span class="keyword">implements</span> <span class="title class_">Iterator</span>&lt;List&lt;Message&gt;&gt; &#123;</span><br><span class="line">   <span class="keyword">private</span> <span class="keyword">final</span> <span class="type">int</span> <span class="variable">SIZE_LIMIT</span> <span class="operator">=</span> <span class="number">1024</span> * <span class="number">1024</span> * <span class="number">4</span>;</span><br><span class="line">   <span class="keyword">private</span> <span class="keyword">final</span> List&lt;Message&gt; messages;</span><br><span class="line">   <span class="keyword">private</span> <span class="type">int</span> currIndex;</span><br><span class="line">   <span class="keyword">public</span> <span class="title function_">ListSplitter</span><span class="params">(List&lt;Message&gt; messages)</span> &#123;</span><br><span class="line">           <span class="built_in">this</span>.messages = messages;</span><br><span class="line">   &#125;</span><br><span class="line">    <span class="meta">@Override</span> </span><br><span class="line">    <span class="keyword">public</span> <span class="type">boolean</span> <span class="title function_">hasNext</span><span class="params">()</span> &#123;</span><br><span class="line">       <span class="keyword">return</span> currIndex &lt; messages.size();</span><br><span class="line">   &#125;</span><br><span class="line">   	<span class="meta">@Override</span> </span><br><span class="line">    <span class="keyword">public</span> List&lt;Message&gt; <span class="title function_">next</span><span class="params">()</span> &#123;</span><br><span class="line">       <span class="type">int</span> <span class="variable">nextIndex</span> <span class="operator">=</span> currIndex;</span><br><span class="line">       <span class="type">int</span> <span class="variable">totalSize</span> <span class="operator">=</span> <span class="number">0</span>;</span><br><span class="line">       <span class="keyword">for</span> (; nextIndex &lt; messages.size(); nextIndex++) &#123;</span><br><span class="line">           <span class="type">Message</span> <span class="variable">message</span> <span class="operator">=</span> messages.get(nextIndex);</span><br><span class="line">           <span class="type">int</span> <span class="variable">tmpSize</span> <span class="operator">=</span> message.getTopic().length() + message.getBody().length;</span><br><span class="line">           Map&lt;String, String&gt; properties = message.getProperties();</span><br><span class="line">           <span class="keyword">for</span> (Map.Entry&lt;String, String&gt; entry : properties.entrySet()) &#123;</span><br><span class="line">               tmpSize += entry.getKey().length() + entry.getValue().length();</span><br><span class="line">           &#125;</span><br><span class="line">           tmpSize = tmpSize + <span class="number">20</span>; <span class="comment">// 增加日志的开销20字节</span></span><br><span class="line">           <span class="keyword">if</span> (tmpSize &gt; SIZE_LIMIT) &#123;</span><br><span class="line">               <span class="comment">//单个消息超过了最大的限制</span></span><br><span class="line">               <span class="comment">//忽略,否则会阻塞分裂的进程</span></span><br><span class="line">               <span class="keyword">if</span> (nextIndex - currIndex == <span class="number">0</span>) &#123;</span><br><span class="line">                  <span class="comment">//假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环</span></span><br><span class="line">                  nextIndex++;</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="keyword">break</span>;</span><br><span class="line">           &#125;</span><br><span class="line">           <span class="keyword">if</span> (tmpSize + totalSize &gt; SIZE_LIMIT) &#123;</span><br><span class="line">               <span class="keyword">break</span>;</span><br><span class="line">           &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">               totalSize += tmpSize;</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">       &#125;</span><br><span class="line">       List&lt;Message&gt; subList = messages.subList(currIndex, nextIndex);</span><br><span class="line">       currIndex = nextIndex;</span><br><span class="line">       <span class="keyword">return</span> subList;</span><br><span class="line">   &#125;</span><br><span class="line">&#125;</span><br><span class="line"><span class="comment">//把大的消息分裂成若干个小的消息</span></span><br><span class="line"><span class="type">ListSplitter</span> <span class="variable">splitter</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">ListSplitter</span>(messages);</span><br><span class="line"><span class="keyword">while</span> (splitter.hasNext()) &#123;</span><br><span class="line">  <span class="keyword">try</span> &#123;</span><br><span class="line">      List&lt;Message&gt;  listItem = splitter.next();</span><br><span class="line">      producer.send(listItem);</span><br><span class="line">  &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">      e.printStackTrace();</span><br><span class="line">      <span class="comment">//处理error</span></span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h2 id="4-5-过滤消息"><a href="#4-5-过滤消息" class="headerlink" title="4.5 过滤消息"></a>4.5 过滤消息</h2><p>在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息。例如：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;CID_EXAMPLE&quot;</span>);</span><br><span class="line">consumer.subscribe(<span class="string">&quot;TOPIC&quot;</span>, <span class="string">&quot;TAGA || TAGB || TAGC&quot;</span>);</span><br></pre></td></tr></table></figure>



<p>消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：</p>
<figure class="highlight text"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line">------------</span><br><span class="line">| message  |</span><br><span class="line">|----------|  a &gt; 5 AND b = &#x27;abc&#x27;</span><br><span class="line">| a = 10   |  --------------------&gt; Gotten</span><br><span class="line">| b = &#x27;abc&#x27;|</span><br><span class="line">| c = true |</span><br><span class="line">------------</span><br><span class="line">------------</span><br><span class="line">| message  |</span><br><span class="line">|----------|   a &gt; 5 AND b = &#x27;abc&#x27;</span><br><span class="line">| a = 1    |  --------------------&gt; Missed</span><br><span class="line">| b = &#x27;abc&#x27;|</span><br><span class="line">| c = true |</span><br><span class="line">------------</span><br></pre></td></tr></table></figure>



<h3 id="4-5-1-SQL基本语法"><a href="#4-5-1-SQL基本语法" class="headerlink" title="4.5.1 SQL基本语法"></a>4.5.1 SQL基本语法</h3><p>RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。</p>
<ul>
<li>数值比较，比如：**&gt;，&gt;&#x3D;，&lt;，&lt;&#x3D;，BETWEEN，&#x3D;；**</li>
<li>字符比较，比如：**&#x3D;，&lt;&gt;，IN；**</li>
<li><strong>IS NULL</strong> 或者 <strong>IS NOT NULL；</strong></li>
<li>逻辑符号 <strong>AND，OR，NOT；</strong></li>
</ul>
<p>常量支持类型为：</p>
<ul>
<li>数值，比如：<strong>123，3.1415；</strong></li>
<li>字符，比如：**’abc’，必须用单引号包裹起来；**</li>
<li><strong>NULL</strong>，特殊的常量</li>
<li>布尔值，<strong>TRUE</strong> 或 <strong>FALSE</strong></li>
</ul>
<p>只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">subscribe</span><span class="params">(finalString topic, <span class="keyword">final</span> MessageSelector messageSelector)</span></span><br></pre></td></tr></table></figure>



<h3 id="4-5-2-消息生产者"><a href="#4-5-2-消息生产者" class="headerlink" title="4.5.2 消息生产者"></a>4.5.2 消息生产者</h3><p>发送消息时，你能通过<code>putUserProperty</code>来设置消息的属性</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="type">DefaultMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQProducer</span>(<span class="string">&quot;please_rename_unique_group_name&quot;</span>);</span><br><span class="line">producer.start();</span><br><span class="line"><span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TopicTest&quot;</span>,</span><br><span class="line">   tag,</span><br><span class="line">   (<span class="string">&quot;Hello RocketMQ &quot;</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET)</span><br><span class="line">);</span><br><span class="line"><span class="comment">// 设置一些属性</span></span><br><span class="line">msg.putUserProperty(<span class="string">&quot;a&quot;</span>, String.valueOf(i));</span><br><span class="line"><span class="type">SendResult</span> <span class="variable">sendResult</span> <span class="operator">=</span> producer.send(msg);</span><br><span class="line"></span><br><span class="line">producer.shutdown();</span><br></pre></td></tr></table></figure>



<h3 id="4-5-3-消息消费者"><a href="#4-5-3-消息消费者" class="headerlink" title="4.5.3 消息消费者"></a>4.5.3 消息消费者</h3><p>用MessageSelector.bySql来使用sql筛选消息</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="type">DefaultMQPushConsumer</span> <span class="variable">consumer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">DefaultMQPushConsumer</span>(<span class="string">&quot;please_rename_unique_group_name_4&quot;</span>);</span><br><span class="line"><span class="comment">// 只有订阅的消息有这个属性a, a &gt;=0 and a &lt;= 3</span></span><br><span class="line">consumer.subscribe(<span class="string">&quot;TopicTest&quot;</span>, MessageSelector.bySql(<span class="string">&quot;a between 0 and 3&quot;</span>);</span><br><span class="line">consumer.registerMessageListener(<span class="keyword">new</span> <span class="title class_">MessageListenerConcurrently</span>() &#123;</span><br><span class="line">   <span class="meta">@Override</span></span><br><span class="line">   <span class="keyword">public</span> ConsumeConcurrentlyStatus <span class="title function_">consumeMessage</span><span class="params">(List&lt;MessageExt&gt; msgs, ConsumeConcurrentlyContext context)</span> &#123;</span><br><span class="line">       <span class="keyword">return</span> ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</span><br><span class="line">   &#125;</span><br><span class="line">&#125;);</span><br><span class="line">consumer.start();</span><br></pre></td></tr></table></figure>



<h2 id="4-6-事务消息"><a href="#4-6-事务消息" class="headerlink" title="4.6 事务消息"></a>4.6 事务消息</h2><p>###4.6.1 流程分析</p>
<p><img  
                     lazyload
                     alt="image"
                     data-src="https://pchaoo.gitee.io/blog/img/rocketmq/%E4%BA%8B%E5%8A%A1%E6%B6%88%E6%81%AF.png"
                      alt="img"
                ></p>
<p>上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。</p>
<p>####1）事务消息发送及提交</p>
<p>(1) 发送消息（half消息）。</p>
<p>(2) 服务端响应消息写入结果。</p>
<p>(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。</p>
<p>(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）</p>
<h4 id="2）事务补偿"><a href="#2）事务补偿" class="headerlink" title="2）事务补偿"></a>2）事务补偿</h4><p>(1) 对没有Commit&#x2F;Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”</p>
<p>(2) Producer收到回查消息，检查回查消息对应的本地事务的状态</p>
<p>(3) 根据本地事务状态，重新Commit或者Rollback</p>
<p>其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。</p>
<h4 id="3）事务消息状态"><a href="#3）事务消息状态" class="headerlink" title="3）事务消息状态"></a>3）事务消息状态</h4><p>事务消息共有三种状态，提交状态、回滚状态、中间状态：</p>
<ul>
<li>TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。</li>
<li>TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。</li>
<li>TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。</li>
</ul>
<p>###4.6.1 发送事务消息</p>
<h4 id="1-创建事务性生产者"><a href="#1-创建事务性生产者" class="headerlink" title="1) 创建事务性生产者"></a>1) 创建事务性生产者</h4><p>使用 TransactionMQProducer类创建生产者，并指定唯一的 ProducerGroup，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">Producer</span> &#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title function_">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> MQClientException, InterruptedException &#123;</span><br><span class="line">        <span class="comment">//创建事务监听器</span></span><br><span class="line">        <span class="type">TransactionListener</span> <span class="variable">transactionListener</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">TransactionListenerImpl</span>();</span><br><span class="line">        <span class="comment">//创建消息生产者</span></span><br><span class="line">        <span class="type">TransactionMQProducer</span> <span class="variable">producer</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">TransactionMQProducer</span>(<span class="string">&quot;group6&quot;</span>);</span><br><span class="line">        producer.setNamesrvAddr(<span class="string">&quot;192.168.25.135:9876;192.168.25.138:9876&quot;</span>);</span><br><span class="line">        <span class="comment">//生产者这是监听器</span></span><br><span class="line">        producer.setTransactionListener(transactionListener);</span><br><span class="line">        <span class="comment">//启动消息生产者</span></span><br><span class="line">        producer.start();</span><br><span class="line">        String[] tags = <span class="keyword">new</span> <span class="title class_">String</span>[]&#123;<span class="string">&quot;TagA&quot;</span>, <span class="string">&quot;TagB&quot;</span>, <span class="string">&quot;TagC&quot;</span>&#125;;</span><br><span class="line">        <span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; <span class="number">3</span>; i++) &#123;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                <span class="type">Message</span> <span class="variable">msg</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">Message</span>(<span class="string">&quot;TransactionTopic&quot;</span>, tags[i % tags.length], <span class="string">&quot;KEY&quot;</span> + i,</span><br><span class="line">                        (<span class="string">&quot;Hello RocketMQ &quot;</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET));</span><br><span class="line">                <span class="type">SendResult</span> <span class="variable">sendResult</span> <span class="operator">=</span> producer.sendMessageInTransaction(msg, <span class="literal">null</span>);</span><br><span class="line">                System.out.printf(<span class="string">&quot;%s%n&quot;</span>, sendResult);</span><br><span class="line">                TimeUnit.SECONDS.sleep(<span class="number">1</span>);</span><br><span class="line">            &#125; <span class="keyword">catch</span> (MQClientException | UnsupportedEncodingException e) &#123;</span><br><span class="line">                e.printStackTrace();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//producer.shutdown();</span></span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h4 id="2）实现事务的监听接口"><a href="#2）实现事务的监听接口" class="headerlink" title="2）实现事务的监听接口"></a>2）实现事务的监听接口</h4><p>当发送半消息成功时，我们使用 <code>executeLocalTransaction</code> 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。<code>checkLocalTranscation</code> 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">TransactionListenerImpl</span> <span class="keyword">implements</span> <span class="title class_">TransactionListener</span> &#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> LocalTransactionState <span class="title function_">executeLocalTransaction</span><span class="params">(Message msg, Object arg)</span> &#123;</span><br><span class="line">        System.out.println(<span class="string">&quot;执行本地事务&quot;</span>);</span><br><span class="line">        <span class="keyword">if</span> (StringUtils.equals(<span class="string">&quot;TagA&quot;</span>, msg.getTags())) &#123;</span><br><span class="line">            <span class="keyword">return</span> LocalTransactionState.COMMIT_MESSAGE;</span><br><span class="line">        &#125; <span class="keyword">else</span> <span class="keyword">if</span> (StringUtils.equals(<span class="string">&quot;TagB&quot;</span>, msg.getTags())) &#123;</span><br><span class="line">            <span class="keyword">return</span> LocalTransactionState.ROLLBACK_MESSAGE;</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            <span class="keyword">return</span> LocalTransactionState.UNKNOW;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> LocalTransactionState <span class="title function_">checkLocalTransaction</span><span class="params">(MessageExt msg)</span> &#123;</span><br><span class="line">        System.out.println(<span class="string">&quot;MQ检查消息Tag【&quot;</span>+msg.getTags()+<span class="string">&quot;】的本地事务执行结果&quot;</span>);</span><br><span class="line">        <span class="keyword">return</span> LocalTransactionState.COMMIT_MESSAGE;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



<h3 id="4-6-2-使用限制"><a href="#4-6-2-使用限制" class="headerlink" title="4.6.2 使用限制"></a>4.6.2 使用限制</h3><ol>
<li>事务消息不支持延时消息和批量消息。</li>
<li>为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N &#x3D; transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionCheckListener 类来修改这个行为。</li>
<li>事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionMsgTimeout 参数。</li>
<li>事务性消息可能不止一次被检查或消费。</li>
<li>提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。</li>
<li>事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。</li>
</ol>

            </div>

            
                <div class="post-copyright-info">
                    
<div class="article-copyright-info-container">
    <ul class="copyright-info-content">
        <li class="post-title">
            <span class="type">本文标题</span>：<span class="content">RocketMQ-01</span>
        </li>
        <li class="post-author">
            <span class="type">本文作者</span>：<span class="content">Blank</span>
        </li>
        <li class="post-time">
            <span class="type">创建时间</span>：<span class="content">2023-02-19 00:00:00</span>
        </li>
        <li class="post-link">
            <span class="type">本文链接</span>：<span class="content">2023/02/19/RocketMQ-01/</span>
        </li>
        <li class="post-license">
            <span class="type">版权声明</span>：<span class="content">本博客所有文章除特别声明外，均采用 <a class="license" target="_blank" rel="noopener" href="https://creativecommons.org/licenses/by-nc-sa/4.0/deed.zh">BY-NC-SA</a> 许可协议。转载请注明出处！</span>
        </li>
    </ul>
    <div class="copy-copyright-info flex-center tooltip" data-content="复制版权信息" data-offset-y="-2px">
        <i class="fa-solid fa-copy"></i>
    </div>
</div>

                </div>
            

            
                <ul class="post-tags-box">
                    
                        <li class="tag-item">
                            <a href="/tags/%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6/">#消息中间件</a>&nbsp;
                        </li>
                    
                </ul>
            

            
                <div class="article-nav">
                    
                        <div class="article-prev">
                            <a class="prev"
                               rel="prev"
                               href="/2023/02/20/MQTT%E5%8D%8F%E8%AE%AE%E4%B8%AD%E6%96%87%E7%89%88/"
                            >
                            <span class="left arrow-icon flex-center">
                              <i class="fas fa-chevron-left"></i>
                            </span>
                                <span class="title flex-center">
                                <span class="post-nav-title-item">MQTT协议中文版</span>
                                <span class="post-nav-item">上一篇</span>
                            </span>
                            </a>
                        </div>
                    
                    
                        <div class="article-next">
                            <a class="next"
                               rel="next"
                               href="/2023/02/19/RocketMQ-02/"
                            >
                            <span class="title flex-center">
                                <span class="post-nav-title-item">RocketMQ-02</span>
                                <span class="post-nav-item">下一篇</span>
                            </span>
                                <span class="right arrow-icon flex-center">
                              <i class="fas fa-chevron-right"></i>
                            </span>
                            </a>
                        </div>
                    
                </div>
            

            
        </div>

        
            <div class="toc-content-container">
                <div class="post-toc-wrap">
    <div class="post-toc">
        <ol class="nav"><li class="nav-item nav-level-1"><a class="nav-link" href="#RocketMQ-01"><span class="nav-number">1.</span> <span class="nav-text">RocketMQ-01</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#1-MQ%E4%BB%8B%E7%BB%8D"><span class="nav-number">2.</span> <span class="nav-text">1. MQ介绍</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#1-2-MQ%E7%9A%84%E4%BC%98%E7%82%B9%E5%92%8C%E7%BC%BA%E7%82%B9"><span class="nav-number">2.1.</span> <span class="nav-text">1.2 MQ的优点和缺点</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#1-3-%E5%90%84%E7%A7%8DMQ%E4%BA%A7%E5%93%81%E7%9A%84%E6%AF%94%E8%BE%83"><span class="nav-number">2.2.</span> <span class="nav-text">1.3 各种MQ产品的比较</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#2-RocketMQ%E5%BF%AB%E9%80%9F%E5%85%A5%E9%97%A8"><span class="nav-number">3.</span> <span class="nav-text">2. RocketMQ快速入门</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#2-1-%E5%87%86%E5%A4%87%E5%B7%A5%E4%BD%9C"><span class="nav-number">3.1.</span> <span class="nav-text">2.1 准备工作</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#2-1-1-%E4%B8%8B%E8%BD%BDRocketMQ"><span class="nav-number">3.1.1.</span> <span class="nav-text">2.1.1 下载RocketMQ</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#2-2-2-%E7%8E%AF%E5%A2%83%E8%A6%81%E6%B1%82"><span class="nav-number">3.1.2.</span> <span class="nav-text">2.2.2 环境要求</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2-2-%E5%AE%89%E8%A3%85RocketMQ"><span class="nav-number">3.2.</span> <span class="nav-text">2.2 安装RocketMQ</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#2-2-1-%E5%AE%89%E8%A3%85%E6%AD%A5%E9%AA%A4"><span class="nav-number">3.2.1.</span> <span class="nav-text">2.2.1 安装步骤</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#2-2-2-%E7%9B%AE%E5%BD%95%E4%BB%8B%E7%BB%8D"><span class="nav-number">3.2.2.</span> <span class="nav-text">2.2.2 目录介绍</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2-3-%E5%90%AF%E5%8A%A8RocketMQ"><span class="nav-number">3.3.</span> <span class="nav-text">2.3 启动RocketMQ</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2-4-%E6%B5%8B%E8%AF%95RocketMQ"><span class="nav-number">3.4.</span> <span class="nav-text">2.4 测试RocketMQ</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#2-4-1-%E5%8F%91%E9%80%81%E6%B6%88%E6%81%AF"><span class="nav-number">3.4.1.</span> <span class="nav-text">2.4.1 发送消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#2-4-2-%E6%8E%A5%E6%94%B6%E6%B6%88%E6%81%AF"><span class="nav-number">3.4.2.</span> <span class="nav-text">2.4.2 接收消息</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2-5-%E5%85%B3%E9%97%ADRocketMQ"><span class="nav-number">3.5.</span> <span class="nav-text">2.5 关闭RocketMQ</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#3-RocketMQ%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA"><span class="nav-number">4.</span> <span class="nav-text">3. RocketMQ集群搭建</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#3-1-%E5%90%84%E8%A7%92%E8%89%B2%E4%BB%8B%E7%BB%8D"><span class="nav-number">4.1.</span> <span class="nav-text">3.1 各角色介绍</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#3-2-%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA%E6%96%B9%E5%BC%8F"><span class="nav-number">4.2.</span> <span class="nav-text">3.2 集群搭建方式</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#3-2-1-%E9%9B%86%E7%BE%A4%E7%89%B9%E7%82%B9"><span class="nav-number">4.2.1.</span> <span class="nav-text">3.2.1 集群特点</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-2-3-%E9%9B%86%E7%BE%A4%E6%A8%A1%E5%BC%8F"><span class="nav-number">4.2.2.</span> <span class="nav-text">3.2.3 集群模式</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89%E5%8D%95Master%E6%A8%A1%E5%BC%8F"><span class="nav-number">4.2.2.1.</span> <span class="nav-text">1）单Master模式</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E5%A4%9AMaster%E6%A8%A1%E5%BC%8F"><span class="nav-number">4.2.2.2.</span> <span class="nav-text">2）多Master模式</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#3%EF%BC%89%E5%A4%9AMaster%E5%A4%9ASlave%E6%A8%A1%E5%BC%8F%EF%BC%88%E5%BC%82%E6%AD%A5%EF%BC%89"><span class="nav-number">4.2.2.3.</span> <span class="nav-text">3）多Master多Slave模式（异步）</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#4%EF%BC%89%E5%A4%9AMaster%E5%A4%9ASlave%E6%A8%A1%E5%BC%8F%EF%BC%88%E5%90%8C%E6%AD%A5%EF%BC%89"><span class="nav-number">4.2.2.4.</span> <span class="nav-text">4）多Master多Slave模式（同步）</span></a></li></ol></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#3-3-%E5%8F%8C%E4%B8%BB%E5%8F%8C%E4%BB%8E%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA"><span class="nav-number">4.3.</span> <span class="nav-text">3.3 双主双从集群搭建</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-1-%E6%80%BB%E4%BD%93%E6%9E%B6%E6%9E%84"><span class="nav-number">4.3.1.</span> <span class="nav-text">3.3.1 总体架构</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-2-%E9%9B%86%E7%BE%A4%E5%B7%A5%E4%BD%9C%E6%B5%81%E7%A8%8B"><span class="nav-number">4.3.2.</span> <span class="nav-text">3.3.2 集群工作流程</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-3-%E6%9C%8D%E5%8A%A1%E5%99%A8%E7%8E%AF%E5%A2%83"><span class="nav-number">4.3.3.</span> <span class="nav-text">3.3.3 服务器环境</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-4-Host%E6%B7%BB%E5%8A%A0%E4%BF%A1%E6%81%AF"><span class="nav-number">4.3.4.</span> <span class="nav-text">3.3.4 Host添加信息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-5-%E9%98%B2%E7%81%AB%E5%A2%99%E9%85%8D%E7%BD%AE"><span class="nav-number">4.3.5.</span> <span class="nav-text">3.3.5 防火墙配置</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-6-%E7%8E%AF%E5%A2%83%E5%8F%98%E9%87%8F%E9%85%8D%E7%BD%AE"><span class="nav-number">4.3.6.</span> <span class="nav-text">3.3.6 环境变量配置</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-7-%E5%88%9B%E5%BB%BA%E6%B6%88%E6%81%AF%E5%AD%98%E5%82%A8%E8%B7%AF%E5%BE%84"><span class="nav-number">4.3.7.</span> <span class="nav-text">3.3.7 创建消息存储路径</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-8-broker%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6"><span class="nav-number">4.3.8.</span> <span class="nav-text">3.3.8 broker配置文件</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89master1"><span class="nav-number">4.3.8.1.</span> <span class="nav-text">1）master1</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89slave2"><span class="nav-number">4.3.8.2.</span> <span class="nav-text">2）slave2</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#3%EF%BC%89master2"><span class="nav-number">4.3.8.3.</span> <span class="nav-text">3）master2</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#4%EF%BC%89slave1"><span class="nav-number">4.3.8.4.</span> <span class="nav-text">4）slave1</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-9-%E4%BF%AE%E6%94%B9%E5%90%AF%E5%8A%A8%E8%84%9A%E6%9C%AC%E6%96%87%E4%BB%B6"><span class="nav-number">4.3.9.</span> <span class="nav-text">3.3.9 修改启动脚本文件</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89runbroker-sh"><span class="nav-number">4.3.9.1.</span> <span class="nav-text">1）runbroker.sh</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-10-%E6%9C%8D%E5%8A%A1%E5%90%AF%E5%8A%A8"><span class="nav-number">4.3.10.</span> <span class="nav-text">3.3.10 服务启动</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89%E5%90%AF%E5%8A%A8NameServe%E9%9B%86%E7%BE%A4"><span class="nav-number">4.3.10.1.</span> <span class="nav-text">1）启动NameServe集群</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E5%90%AF%E5%8A%A8Broker%E9%9B%86%E7%BE%A4"><span class="nav-number">4.3.10.2.</span> <span class="nav-text">2）启动Broker集群</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-11-%E6%9F%A5%E7%9C%8B%E8%BF%9B%E7%A8%8B%E7%8A%B6%E6%80%81"><span class="nav-number">4.3.11.</span> <span class="nav-text">3.3.11 查看进程状态</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-3-12-%E6%9F%A5%E7%9C%8B%E6%97%A5%E5%BF%97"><span class="nav-number">4.3.12.</span> <span class="nav-text">3.3.12 查看日志</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#3-4-mqadmin%E7%AE%A1%E7%90%86%E5%B7%A5%E5%85%B7"><span class="nav-number">4.4.</span> <span class="nav-text">3.4 mqadmin管理工具</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#3-4-1-%E4%BD%BF%E7%94%A8%E6%96%B9%E5%BC%8F"><span class="nav-number">4.4.1.</span> <span class="nav-text">3.4.1 使用方式</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#5%EF%BC%89%E6%B6%88%E8%B4%B9%E8%80%85%E3%80%81%E6%B6%88%E8%B4%B9%E7%BB%84%E7%9B%B8%E5%85%B3"><span class="nav-number">4.4.1.1.</span> <span class="nav-text">5）消费者、消费组相关</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#6%EF%BC%89%E8%BF%9E%E6%8E%A5%E7%9B%B8%E5%85%B3"><span class="nav-number">4.4.1.2.</span> <span class="nav-text">6）连接相关</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#7%EF%BC%89NameServer%E7%9B%B8%E5%85%B3"><span class="nav-number">4.4.1.3.</span> <span class="nav-text">7）NameServer相关</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#8%EF%BC%89%E5%85%B6%E4%BB%96"><span class="nav-number">4.4.1.4.</span> <span class="nav-text">8）其他</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-4-3-%E6%B3%A8%E6%84%8F%E4%BA%8B%E9%A1%B9"><span class="nav-number">4.4.2.</span> <span class="nav-text">3.4.3 注意事项</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#3-5-%E9%9B%86%E7%BE%A4%E7%9B%91%E6%8E%A7%E5%B9%B3%E5%8F%B0%E6%90%AD%E5%BB%BA"><span class="nav-number">4.5.</span> <span class="nav-text">3.5 集群监控平台搭建</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#3-5-1-%E6%A6%82%E8%BF%B0"><span class="nav-number">4.5.1.</span> <span class="nav-text">3.5.1 概述</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-5-2-%E4%B8%8B%E8%BD%BD%E5%B9%B6%E7%BC%96%E8%AF%91%E6%89%93%E5%8C%85"><span class="nav-number">4.5.2.</span> <span class="nav-text">3.5.2 下载并编译打包</span></a></li></ol></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#4-%E6%B6%88%E6%81%AF%E5%8F%91%E9%80%81%E6%A0%B7%E4%BE%8B"><span class="nav-number">5.</span> <span class="nav-text">4. 消息发送样例</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#4-1-%E5%9F%BA%E6%9C%AC%E6%A0%B7%E4%BE%8B"><span class="nav-number">5.1.</span> <span class="nav-text">4.1 基本样例</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#4-1-1-%E6%B6%88%E6%81%AF%E5%8F%91%E9%80%81"><span class="nav-number">5.1.1.</span> <span class="nav-text">4.1.1 消息发送</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89%E5%8F%91%E9%80%81%E5%90%8C%E6%AD%A5%E6%B6%88%E6%81%AF"><span class="nav-number">5.1.1.1.</span> <span class="nav-text">1）发送同步消息</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E5%8F%91%E9%80%81%E5%BC%82%E6%AD%A5%E6%B6%88%E6%81%AF"><span class="nav-number">5.1.1.2.</span> <span class="nav-text">2）发送异步消息</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#3%EF%BC%89%E5%8D%95%E5%90%91%E5%8F%91%E9%80%81%E6%B6%88%E6%81%AF"><span class="nav-number">5.1.1.3.</span> <span class="nav-text">3）单向发送消息</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-1-2-%E6%B6%88%E8%B4%B9%E6%B6%88%E6%81%AF"><span class="nav-number">5.1.2.</span> <span class="nav-text">4.1.2 消费消息</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1%EF%BC%89%E8%B4%9F%E8%BD%BD%E5%9D%87%E8%A1%A1%E6%A8%A1%E5%BC%8F"><span class="nav-number">5.1.2.1.</span> <span class="nav-text">1）负载均衡模式</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E5%B9%BF%E6%92%AD%E6%A8%A1%E5%BC%8F"><span class="nav-number">5.1.2.2.</span> <span class="nav-text">2）广播模式</span></a></li></ol></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#4-2-%E9%A1%BA%E5%BA%8F%E6%B6%88%E6%81%AF"><span class="nav-number">5.2.</span> <span class="nav-text">4.2 顺序消息</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#4-2-1-%E9%A1%BA%E5%BA%8F%E6%B6%88%E6%81%AF%E7%94%9F%E4%BA%A7"><span class="nav-number">5.2.1.</span> <span class="nav-text">4.2.1 顺序消息生产</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-2-2-%E9%A1%BA%E5%BA%8F%E6%B6%88%E8%B4%B9%E6%B6%88%E6%81%AF"><span class="nav-number">5.2.2.</span> <span class="nav-text">4.2.2 顺序消费消息</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#4-3-%E5%BB%B6%E6%97%B6%E6%B6%88%E6%81%AF"><span class="nav-number">5.3.</span> <span class="nav-text">4.3 延时消息</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#4-3-1-%E5%90%AF%E5%8A%A8%E6%B6%88%E6%81%AF%E6%B6%88%E8%B4%B9%E8%80%85"><span class="nav-number">5.3.1.</span> <span class="nav-text">4.3.1 启动消息消费者</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-3-2-%E5%8F%91%E9%80%81%E5%BB%B6%E6%97%B6%E6%B6%88%E6%81%AF"><span class="nav-number">5.3.2.</span> <span class="nav-text">4.3.2 发送延时消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-3-4-%E4%BD%BF%E7%94%A8%E9%99%90%E5%88%B6"><span class="nav-number">5.3.3.</span> <span class="nav-text">4.3.4 使用限制</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#4-4-%E6%89%B9%E9%87%8F%E6%B6%88%E6%81%AF"><span class="nav-number">5.4.</span> <span class="nav-text">4.4 批量消息</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#4-4-1-%E5%8F%91%E9%80%81%E6%89%B9%E9%87%8F%E6%B6%88%E6%81%AF"><span class="nav-number">5.4.1.</span> <span class="nav-text">4.4.1 发送批量消息</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#4-5-%E8%BF%87%E6%BB%A4%E6%B6%88%E6%81%AF"><span class="nav-number">5.5.</span> <span class="nav-text">4.5 过滤消息</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#4-5-1-SQL%E5%9F%BA%E6%9C%AC%E8%AF%AD%E6%B3%95"><span class="nav-number">5.5.1.</span> <span class="nav-text">4.5.1 SQL基本语法</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-5-2-%E6%B6%88%E6%81%AF%E7%94%9F%E4%BA%A7%E8%80%85"><span class="nav-number">5.5.2.</span> <span class="nav-text">4.5.2 消息生产者</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-5-3-%E6%B6%88%E6%81%AF%E6%B6%88%E8%B4%B9%E8%80%85"><span class="nav-number">5.5.3.</span> <span class="nav-text">4.5.3 消息消费者</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#4-6-%E4%BA%8B%E5%8A%A1%E6%B6%88%E6%81%AF"><span class="nav-number">5.6.</span> <span class="nav-text">4.6 事务消息</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E4%BA%8B%E5%8A%A1%E8%A1%A5%E5%81%BF"><span class="nav-number">5.6.0.1.</span> <span class="nav-text">2）事务补偿</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#3%EF%BC%89%E4%BA%8B%E5%8A%A1%E6%B6%88%E6%81%AF%E7%8A%B6%E6%80%81"><span class="nav-number">5.6.0.2.</span> <span class="nav-text">3）事务消息状态</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-%E5%88%9B%E5%BB%BA%E4%BA%8B%E5%8A%A1%E6%80%A7%E7%94%9F%E4%BA%A7%E8%80%85"><span class="nav-number">5.6.0.3.</span> <span class="nav-text">1) 创建事务性生产者</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2%EF%BC%89%E5%AE%9E%E7%8E%B0%E4%BA%8B%E5%8A%A1%E7%9A%84%E7%9B%91%E5%90%AC%E6%8E%A5%E5%8F%A3"><span class="nav-number">5.6.0.4.</span> <span class="nav-text">2）实现事务的监听接口</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-6-2-%E4%BD%BF%E7%94%A8%E9%99%90%E5%88%B6"><span class="nav-number">5.6.1.</span> <span class="nav-text">4.6.2 使用限制</span></a></li></ol></li></ol></li></ol>
    </div>
</div>

            </div>
        
    </div>
</div>


                
            </div>

        </div>

        <div class="page-main-content-bottom">
            
<footer class="footer">
    <div class="info-container">
        <div class="copyright-info info-item">
            &copy;
            
            2024
            
                &nbsp;<i class="fas fa-heart icon-animate"></i>
                &nbsp;<a href="/">Blank</a>
            
        </div>
        
            <script async data-pjax
                    src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script>
            <div class="website-count info-item">
                
                
            </div>
        
        <div class="theme-info info-item">
            由 <a target="_blank" href="https://hexo.io">Hexo</a> 驱动&nbsp;|&nbsp;主题&nbsp;<a class="theme-version" target="_blank" href="https://github.com/XPoet/hexo-theme-keep">Keep v3.6.1</a>
        </div>
        
        
            <div class="deploy-info info-item">
                
                    <a target="_blank" rel="nofollow" href="https://gitee.com/lucky0915">
                
                    本站由 <span class="tooltip" data-content="Gitee Pages"><img src="/images/deploy-provider/gitee.png"></span> 提供部署服务
                
                    </a>
                
            </div>
        
    </div>
</footer>

        </div>
    </div>

    
        <div class="post-tools">
            <div class="post-tools-container">
    <ul class="tools-list">
        <!-- TOC aside toggle -->
        
            <li class="tools-item flex-center toggle-show-toc">
                <i class="fas fa-list"></i>
            </li>
        

        <!-- go comment -->
        
    </ul>
</div>

        </div>
    

    <div class="right-bottom-side-tools">
        <div class="side-tools-container">
    <ul class="side-tools-list">
        <li class="tools-item tool-font-adjust-plus flex-center">
            <i class="fas fa-search-plus"></i>
        </li>

        <li class="tools-item tool-font-adjust-minus flex-center">
            <i class="fas fa-search-minus"></i>
        </li>

        <li class="tools-item tool-dark-light-toggle flex-center">
            <i class="fas fa-moon"></i>
        </li>

        <!-- rss -->
        

        

        <li class="tools-item tool-scroll-to-bottom flex-center">
            <i class="fas fa-arrow-down"></i>
        </li>
    </ul>

    <ul class="exposed-tools-list">
        <li class="tools-item tool-toggle-show flex-center">
            <i class="fas fa-cog fa-spin"></i>
        </li>
        
            <li class="tools-item tool-scroll-to-top flex-center">
                <i class="arrow-up fas fa-arrow-up"></i>
                <span class="percent"></span>
            </li>
        
    </ul>
</div>

    </div>

    <div class="zoom-in-image-mask">
    <img class="zoom-in-image">
</div>


    
        <div class="search-pop-overlay">
    <div class="popup search-popup">
        <div class="search-header">
          <span class="search-input-field-pre">
            <i class="fas fa-keyboard"></i>
          </span>
            <div class="search-input-container">
                <input autocomplete="off"
                       autocorrect="off"
                       autocapitalize="off"
                       placeholder="搜索..."
                       spellcheck="false"
                       type="search"
                       class="search-input"
                >
            </div>
            <span class="close-popup-btn">
                <i class="fas fa-times"></i>
            </span>
        </div>
        <div id="search-result">
            <div id="no-result">
                <i class="fas fa-spinner fa-pulse fa-5x fa-fw"></i>
            </div>
        </div>
    </div>
</div>

    

</main>



<script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/utils.js"></script><script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/main.js"></script><script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/header-shrink.js"></script><script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/back2top.js"></script><script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/dark-light-toggle.js"></script>




    <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/local-search.js"></script>



    <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/code-block.js"></script>



    <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/lazyload.js"></script>


<div class="post-scripts pjax">
    
        <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/post-helper.js"></script>
        
            <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/libs/anime.min.js"></script>
        
        
            <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/toc.js"></script>
        
    
</div>


    <script src="//cdn.jsdelivr.net/npm/hexo-theme-keep@3.6.1/source/js/libs/pjax.min.js"></script>
<script>
    window.addEventListener('DOMContentLoaded', () => {
        window.pjax = new Pjax({
            selectors: [
                'head title',
                '.page-container',
                '.pjax'
            ],
            history: true,
            debug: false,
            cacheBust: false,
            timeout: 0,
            analytics: false,
            currentUrlFullReload: false,
            scrollRestoration: false,
            // scrollTo: true,
        });

        document.addEventListener('pjax:send', () => {
            KEEP.utils.pjaxProgressBarStart();
        });

        document.addEventListener('pjax:complete', () => {
            KEEP.utils.pjaxProgressBarEnd();
            window.pjax.executeScripts(document.querySelectorAll('script[data-pjax], .pjax script'));
            KEEP.refresh();
        });
    });
</script>



</body>
</html>
